eivintobias/heartly-qwen-code
1
0
Fork 0
Code Issues Pull Requests Actions Projects Releases Wiki Activity

初始化项目，由ModelHub XC社区提供模型

Browse Source 
Model: eivintobias/heartly-qwen-code
Source: Original Platform
This commit is contained in:
ModelHub XC
2026-09-04 04:10:21 +08:00
commit dd9551524a
12 changed files with 973 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

452
reply_formatter.py Normal file
View File

@@ -0,0 +1,452 @@
#!/usr/bin/env python3
"""
reply_formatter.py -- extract the user-visible answer from a Heartly
Qwen-Code (v3) grammar output.
The v3 model is trained to emit:
thinking {reasoning} response<decide>speak|stop</decide><verify>known|unknown</verify> {answer} <stop>
(or `<decide>stop</decide>` for the silence case).
The user should only ever see {answer}, never the scaffolding. In practice the
model -- especially through a GUI like LM Studio -- produces *noisy* variants:
the tags are NOT registered as tokenizer special tokens, so Qwen's BPE shatters
`<decide>` into subwords that can re-decode mangled (e.g. ``<deside>``), the
`<stop>` end-marker can come back truncated (no closing `>`), and verify values
can carry junk (`<verify>known, true</verify>`). Qwen chat-template tokens also
leak as plain text (`<stop> <begin> <sep>`).
This parser is tolerant: it canonicalises the tags, extracts the structured
fields, and has an aggressive fallback so NO grammar token ever reaches the user.
Modes:
"chat" -- only the clean answer (default)
"debug" -- answer + [decide=X verify=Y] metadata per turn
"raw" -- the original raw output, untouched
Usage:
from reply_formatter import format_reply, clean_reply
shown = format_reply(raw_model_output)
"""
from __future__ import annotations
import re
from dataclasses import dataclass
# ---------------------------------------------------------------------------
# 1. Junk tokens -- decoded-to-plain-text Qwen special tokens that are never
# answer content. Stripped first so they cannot confuse grammar parsing.
# ---------------------------------------------------------------------------
_JUNK_TOKENS = (
"<|im_start|>", "<|im_end|>", "im_start", "im_end",
"<|object_ref_start|>", "<|object_ref_end|>",
"<|box_start|>", "<|box_end|>",
"<|quad_start|>", "<|quad_end|>",
"<|vision_start|>", "<|vision_end|>", "<|vision_pad|>", "<|image_pad|>",
"<|endoftext|>", "endoftext",
"<|tool_call_begin|>", "<|tool_call_end|>", "<|tool_call_argument_begin|>",
"<|tool_call_argument_end|>", "<|tool_call_argument_name|>", "<|tool_call_argument|>",
"<|tool_calls_section_begin|>", "<|tool_calls_section_end|>",
"<tool_calls>", "</tool_calls>", "tool_call", "tool_calls",
"<begin>", "<sep>",
)
# ---------------------------------------------------------------------------
# 2. Meta-commentary patterns -- the model's self-talk about what it knows /
# doesn't know. If any bleeds into the answer zone, strip it. Sourced from
# the reasoning / refusal / silence templates in render_code_sft_v3.py.
# (The think block is removed entirely in the common case, so this is a
# safety net, not the primary mechanism.)
# ---------------------------------------------------------------------------
_META_PATTERNS = [
r"\bI (?:do )?know (?:how )?to do this\b[^.]*\.",
r"\bI (?:do )?know how to write this\b[^.]*\.",
r"\bI (?:do )?know this function well\b[^.]*\.",
r"\bI (?:can|will) (?:write|produce|implement|respond|answer|write clean code)\b[^.]*\.",
r"\bThis is a standard programming task\b[^.]*\.",
r"\bI recognise this programming problem\b[^.]*\.",
r"\bI (?:recognise|recognize) this (?:programming|pattern|algorithm)\b[^.]*\.",
r"\bStandard problem\b[^.]*\.",
r"\bClear task\b[^.]*\.",
r"\bI've seen this pattern before\b[^.]*\.",
r"\bThis is straightforward\b[^.]*\.",
r"\bI (?:do )?not know this API or library\b[^.]*\.",
r"\bI should not invent a solution\b[^.]*\.",
r"\bI have no knowledge of this framework\b[^.]*\.",
r"\bThe honest response is\b[^.]*\.",
r"\bI cannot verify the correct implementation\b[^.]*\.",
r"\bMaking something up would be worse than admitting it\b[^.]*\.",
r"\bI'll say so rather than produce fake code\b[^.]*\.",
r"\bThe input is empty or not a real question\b[^.]*\.",
r"\bNo meaningful request was made\b[^.]*\.",
r"\bSpeaking would add nothing\b[^.]*\.",
r"\bSocial turn\b[^.]*\.",
r"\bGreeting[ —-]\s*respond\b[^.]*\.",
r"\bNot a factual question\b[^.]*\.",
r"\bCasual conversation\b[^.]*\.",
r"\bFollow-up question\b[^.]*\.",
r"\bThey need motivation\b[^.]*\.",
r"\bEmotional context\b[^.]*\.",
r"\bMeta-conversation about how we work together\b[^.]*\.",
r"\bBe warm\b[^.]*\.",
r"\bThey (?:want|are|need|'re|'ve|gave) ",
r"\bThey're (?:just saying|opening up) ",
r"\bThey gave more detail\b[^.]*\.",
]
_META_RE = re.compile("|".join(_META_PATTERNS), re.IGNORECASE)
# Stray control words left behind after a mangled tag is stripped.
# Any leftover angle-bracket control construct, tolerant of truncation/mangles
# (e.g. `</decide` with no closing `>`, or `<deside speak>`).
_TAG_RE = re.compile(r"</?\s*[a-z_][^>]*>?", re.IGNORECASE)
# Stray control words left behind after a mangled tag is stripped (used inline
# in _clean_text as leading/trailing guards).
_CONTROL_WORDS = r"(?:speak|stop|silent|done|response|known|unknown|precise)"
# Collapse runs of whitespace.
_MULTI_SPACE_RE = re.compile(r"\s{2,}")
# ---------------------------------------------------------------------------
# 3. Normalization -- turn messy real-world tokens into canonical tags so the
# structured parser can run cleanly.
# ---------------------------------------------------------------------------
def _normalize(raw: str) -> str:
t = raw or ""
# 3a. Drop Qwen / decoded-token litter.
for junk in _JUNK_TOKENS:
t = t.replace(junk, "")
# 3b. Decide tags. Canonical is ``<decide>`` / ``</decide>`` but tokenization
# can drop a letter (<deside>) or add spaces, and the close can be
# truncated (``</decide``). ``de[cs]ide`` catches both spellings.
t = re.sub(
r"<\s*/?\s*de[cs]ide\b[^>]*>?>?",
lambda m: "</decide>" if "/" in m.group(0) else "<decide>",
t,
flags=re.IGNORECASE,
)
# 3c. Verify tags -- same tolerance.
t = re.sub(
r"<\s*/?\s*verify\b[^>]*>?>?",
lambda m: "</verify>" if "/" in m.group(0) else "<verify>",
t,
flags=re.IGNORECASE,
)
# 3d. Stop tags -- ``<stop>``, ``<stop (truncated`` (no close), and a bare
# ``<stop`` at end-of-output. The optional ``>?>`` handles all of these.
t = re.sub(r"<\s*stop\b[^>]*>?>?", "<stop>", t, flags=re.IGNORECASE)
# 3e. Strip a leading noise run (``<stop> <begin> <sep>`` ...) that some
# outputs carry before any thinking/decide content. This <stop> is junk,
# not the Heartly end-of-turn marker (which is mid/trailing).
t = re.sub(
r"^\s*((?:<stop>|<begin>|<sep>)\s*)*", "", t, flags=re.IGNORECASE
).lstrip()
return t
# ---------------------------------------------------------------------------
# 4. Value normalization helpers.
# ---------------------------------------------------------------------------
_DECIDE_VAL_RE = re.compile(r"\b(speak|stop|silent)\b", re.IGNORECASE)
def _decide_value(raw_val: str) -> str:
"""Reduce a decide tag body to canonical ``speak`` / ``stop``."""
m = _DECIDE_VAL_RE.search(raw_val or "")
if m:
v = m.group(1).lower()
return "stop" if v == "silent" else v
return "" # unparseable
_VERIFY_VAL_RE = re.compile(r"\b(known|unknown)\b", re.IGNORECASE)
def _verify_value(raw_val: str) -> str:
m = _VERIFY_VAL_RE.search(raw_val or "")
return m.group(1).lower() if m else ""
# ---------------------------------------------------------------------------
# 5. Parsed turn + segment parser.
# ---------------------------------------------------------------------------
@dataclass
class ParsedTurn:
reasoning: str = ""
decide: str = "" # speak | stop | ""
verify: str = "" # known | unknown | ""
answer: str = ""
# ``thinking {reasoning} response<decide>...`` -- the word "response" is
# OPTIONAL (the model drops it sometimes; seen in the 2026-08-01 log). The
# lookahead ``(?=<decide>|\\Z)`` bounds the reasoning block at the decide tag
# (or end of string) so the (non-greedy) body doesn't greedily swallow the
# answer, and it does NOT consume the decide opener (leaving it for the decide
# parser to read the decide value).
_THINK_RE = re.compile(
r"\bthinking\b\s*(.*?)(\s*\bresponse\b)?\s*(?=<decide>|\Z)",
re.DOTALL | re.IGNORECASE,
)
# After normalization, tags are canonical.
_DECIDE_CLOSE_RE = re.compile(r"<decide>(.*?)</decide>", re.DOTALL | re.IGNORECASE)
_VERIFY_CLOSE_RE = re.compile(r"<verify>(.*?)</verify>", re.DOTALL | re.IGNORECASE)
# Tolerant open-tag grabs (for mangled/missing close).
_DECIDE_OPEN_RE = re.compile(r"<decide\b([^<]*)", re.IGNORECASE)
_VERIFY_OPEN_RE = re.compile(r"<verify\b([^<]*)", re.IGNORECASE)
def _parse_segment(segment: str) -> ParsedTurn:
"""Parse one turn (already split on ``<stop>``)."""
seg = segment.strip()
if not seg:
return ParsedTurn()
# --- thinking block (word "response" optional) ---
reasoning = ""
m = _THINK_RE.search(seg)
if m:
reasoning = (m.group(1) or "").strip()
seg = seg[m.end():]
# --- decide ---
decide = ""
m = _DECIDE_CLOSE_RE.search(seg)
if m:
decide = _decide_value(m.group(1))
seg = seg[m.end():]
else:
mo = _DECIDE_OPEN_RE.search(seg)
if mo:
decide = _decide_value(mo.group(1))
seg = seg[mo.end():]
# --- verify ---
verify = ""
m = _VERIFY_CLOSE_RE.search(seg)
if m:
verify = _verify_value(m.group(1))
seg = seg[m.end():]
else:
mo = _VERIFY_OPEN_RE.search(seg)
if mo:
verify = _verify_value(mo.group(1))
seg = seg[mo.end():]
# --- answer zone: everything left, up to any stray stop ---
answer = seg
sm = re.search(r"<stop", answer, re.IGNORECASE)
if sm:
answer = answer[: sm.start()]
# Default an unparseable decide to "speak" (the model is answering).
if not decide:
decide = "speak"
return ParsedTurn(
reasoning=reasoning,
decide=decide,
verify=verify,
answer=answer,
)
def parse_reply(raw: str) -> list[ParsedTurn]:
"""Parse raw model output into one or more turns (multi-turn aware)."""
text = _normalize(raw)
turns: list[ParsedTurn] = []
for segment in re.split(r"<stop>", text):
seg = segment.strip()
if not seg:
continue
# A real turn must carry a decide marker; stray fence/backtick noise
# emitted between repeated <stop> markers is skipped.
if not _DECIDE_OPEN_RE.search(seg):
continue
turns.append(_parse_segment(seg))
if not turns:
# No structured grammar found at all -- treat the whole output as one
# best-effort turn.
turns.append(_parse_segment(text))
return turns
# ---------------------------------------------------------------------------
# 6. Answer sanitizers.
# ---------------------------------------------------------------------------
def _strip_meta(text: str) -> str:
"""Remove the model's reasoning self-talk from a piece of text."""
return _META_RE.sub("", text).strip()
def _dedupe(text: str) -> str:
"""Drop verbatim-repeated sentences, keep first occurrence."""
out, seen = [], set()
for sent in re.split(r"(?<=[.!?])\s+", text):
s = sent.strip()
if not s:
continue
key = s.lower()
if key not in seen:
seen.add(key)
out.append(s)
return " ".join(out)
def _clean_text(text: str) -> str:
"""Final pass on an answer zone: strip residual tags/junk, meta, dedupe.
Whitespace collapse and sentence de-duplication run on the PROSE regions
only -- fenced code blocks (``` ``` ```) are preserved verbatim, because
their indentation is significant and must not be collapsed to a single
space.
"""
t = text or ""
# Qwen litter first (never answer content).
for junk in _JUNK_TOKENS:
t = t.replace(junk, "")
# Fence out code blocks so their whitespace survives intact. re.split with
# a capturing group yields [prose, code, prose, code, ...] -- code blocks
# land at odd indices and are emitted untouched.
pieces = re.split(r"(```[^\n]*\n.*?```)", t, flags=re.DOTALL)
out = []
for i, chunk in enumerate(pieces):
if not chunk:
continue
if i % 2 == 1: # captured fenced block -> keep verbatim
# The model emits literal backslash-n / backslash-t as plain text in
# code (training artifact). Restore real newlines/tabs so multi-line
# code renders instead of collapsing to a single line.
chunk = chunk.replace(chr(92) + "n", chr(10)).replace(chr(92) + "t", chr(9))
out.append(chunk)
continue
# prose region
chunk = _TAG_RE.sub(" ", chunk)
chunk = re.sub(rf"^\s*{_CONTROL_WORDS}\b\s*", "", chunk, flags=re.IGNORECASE)
chunk = re.sub(rf"\s*{_CONTROL_WORDS}\s*$", "", chunk, flags=re.IGNORECASE)
chunk = _strip_meta(chunk)
chunk = _MULTI_SPACE_RE.sub(" ", chunk).strip()
chunk = _dedupe(chunk)
if chunk:
out.append(chunk)
return " ".join(out).strip()
def _answer_from_reasoning(reasoning: str) -> str:
"""Last-resort: if the answer zone is empty, try to harvest a real answer
from the reasoning block (e.g. 'The answer is X' phrasing). Returns "" when
the reasoning only holds placeholder / meta chatter -- in that case the
caller maps the empty answer to the silence sentinel ("...")."""
if not reasoning:
return ""
m = re.search(
r"(?:the answer is|answer is|it's|it is)\s*:?\s*(.+?)(?:\.\s*$|\.\s*<|$)",
reasoning,
re.IGNORECASE | re.DOTALL,
)
if m:
return _clean_text(m.group(1))
return ""
# ---------------------------------------------------------------------------
# 7. Resolution -- decide what the user actually sees.
# ---------------------------------------------------------------------------
def _resolve_chat(turns: list[ParsedTurn]) -> str:
parts = []
for t in turns:
if t.decide == "stop":
parts.append("...")
continue
ans = _clean_text(t.answer)
if not ans and t.verify == "unknown":
parts.append(ans or "I don't have that information.")
continue
if not ans:
ans = _answer_from_reasoning(t.reasoning)
parts.append(ans if ans else "...")
return "\n".join(parts)
def _resolve_debug(turns):
parts = []
for t in turns:
ans = _clean_text(t.answer)
if not ans:
ans = _answer_from_reasoning(t.reasoning)
meta = " ".join(
f"[{k}={v}]" for k, v in (("decide", t.decide), ("verify", t.verify)) if v
)
shown = f"{ans} {meta}".strip() if ans else (meta or "(empty)")
parts.append(shown)
return "\n".join(parts)
# 6b. Last-resort legacy cleaner -- used only if structured parsing leaves a
# grammar token in the chat output. Aggressively strips every angle-bracket
# construct + control word, guaranteeing a clean result.
def _legacy_clean(text):
t = _normalize(text)
t = re.split(r"<stop", t, maxsplit=1)[0]
dm = _DECIDE_OPEN_RE.search(t)
if dm:
t = t[dm.start():]
else:
m = _THINK_RE.search(t)
if m:
t = t[m.end():] if m.end() <= len(t) else ""
t = _DECIDE_CLOSE_RE.sub(" ", t)
t = _VERIFY_CLOSE_RE.sub(" ", t)
t = _TAG_RE.sub(" ", t)
t = re.sub(rf"^\s*{_CONTROL_WORDS}\b\s*", "", t, flags=re.IGNORECASE)
t = re.sub(rf"\s*{_CONTROL_WORDS}\b\s*$", "", t, flags=re.IGNORECASE)
return _clean_text(t)
# 8. Public API
def format_reply(raw, mode="chat"):
"""Format a raw Heartly model output for display.
mode: "chat" (default), "debug", or "raw".
"""
if mode == "raw":
return raw if raw else ""
if mode not in ("chat", "debug"):
mode = "chat"
if not raw or not raw.strip():
return ""
turns = parse_reply(raw)
if mode == "debug":
return _resolve_debug(turns)
out = _resolve_chat(turns)
# Leak guarantee: if any grammar token survived, fall back hard.
if out == "" and raw.strip():
out = _legacy_clean(raw)
elif out != "" and _TAG_RE.search(out):
out = _legacy_clean(raw) or out
# Safety net: the model intermittently emits code (especially un-fenced) as
# literal backslash-n / backslash-t text. Render those as real newlines/tabs
# so answers never collapse to a single line in the browser or CLI.
out = out.replace(chr(92) + "n", chr(10)).replace(chr(92) + "t", chr(9))
return out
def clean_reply(text):
"""Legacy-compatible drop-in replacement for the old clean_reply()."""
return format_reply(text, mode="chat")
if __name__ == "__main__":
import sys
sample = sys.stdin.read() if not sys.argv[1:] else open(sys.argv[1]).read()
print(format_reply(sample, mode="debug"))