"source": "VOCAB = list('{}\",:[]abcdefghijklmnopqrstuvwxyz0123456789_. -') + [\"\"]\n\nclass SchemaFSM:\n '''Enforces: {\"name\": \"\", \"age\": , \"active\": } (in that order).\n\n Each state returns the SET of characters legal as the very next character.\n That set is exactly the mask a serving engine applies to the logits.\n '''\n def __init__(self):\n self.reset()\n\n def reset(self):\n self.buf = \"\"\n self.state = \"start\"\n self.field = 0\n return self\n\n FIELDS = [(\"name\", \"string\"), (\"age\", \"int\"), (\"active\", \"bool\")]\n\n def allowed(self):\n s, f = self.state, self.field\n if s == \"start\": return {\"{\"}\n if s == \"pre_key\": return {'\"'}\n if s == \"in_key\": # only the exact expected key\n key = self.FIELDS[f][0]\n done = self.buf\n return {key[len(done)]} if len(done) < len(key) else {'\"'}\n if s == \"post_key\": return {\":\"}\n if s == \"pre_val\":\n kind = self.FIELDS[f][1]\n if kind == \"string\": return {'\"'}\n if kind == \"int\": return set(\"123456789\") # no leading zero\n if kind == \"bool\": return {\"t\", \"f\"}\n if s == \"in_string\":\n # maxLength=24, exactly like a JSON-schema bound: past it, only the closing quote is legal\n return {'\"'} if len(self.buf) >= 24 else set(\"abcdefghijklmnopqrstuvwxyz _-\") | {'\"'}\n if s == \"in_int\":\n close = {\",\"} if f < 2 else {\"}\"}\n return close if len(self.buf) >= 5 else set(\"0123456789\") | close\n if s == \"in_bool\":\n word = \"true\" if self.buf[0] == \"t\" else \"false\"\n return {word[len(self.buf)]} if len(self.buf) < len(word) else ({\",\"} if f < 2 else {\"}\"})\n if s == \"post_val\": return {\",\"} if f < len(self.FIELDS) - 1 else {\"}\"}\n if s == \"done\": return {\"\"}\n return set()\n\n def step(self, ch):\n s = self.state\n if s == \"start\" and ch == \"{\": self.state, self.buf = \"pre_key\", \"\"\n elif s == \"pre_key\" and ch == '\"': self.state, self.buf = \"in_key\", \"\"\n elif s == \"in_key\":\n if ch == '\"': self.state = \"post_key\"\n else: self.buf += ch\n elif s == \"post_key\" and ch == \":\": self.state, self.buf = \"pre_val\", \"\"\n elif s == \"pre_val\":\n kind = self.FIELDS[self.field][1]\n if kind == \"string\" and ch == '\"': self.state, self.buf = \"in_string\", \"\"\n elif kind == \"int\": self.state, self.buf = \"in_int\", ch\n elif kind == \"bool\": self.state, self.buf = \"in_bool\", ch\n elif s == \"in_string\":\n if ch == '\"': self.state = \"post_val\"\n else: self.buf += ch\n elif s == \"in_int\":\n if ch in \",}\": self.state = \"post_val\"; return self.step(ch)\n else: self.buf += ch\n elif s == \"in_bool\":\n word = \"true\" if self.buf[0] == \"t\" else \"false\"\n if len(self.buf) < len(word): self.buf += ch\n if self.buf == word and ch in \",}\": self.state = \"post_val\"; return self.step(ch)\n elif s == \"post_val\":\n if ch == \",\": self.state = \"pre_key\"; self.field += 1\n elif ch == \"}\": self.state = \"done\"\n return self\n\n# Drive it with a model that outputs PURE NOISE - if the FSM works, the output is still valid JSON.\ndef generate_masked(fsm, rng, max_chars=120):\n fsm.reset(); out = []\n for _ in range(max_chars):\n allowed = fsm.allowed()\n if allowed == {\"\"}: break\n ch = rng.choice(sorted(allowed)) # a random pick among LEGAL characters only\n out.append(ch); fsm.step(ch)\n return \"\".join(out)\n\nprint(\"Three samples from a *uniformly random* generator, constrained by the FSM:\\n\")\nfor i in range(3):\n text = generate_masked(SchemaFSM(), random.Random(i))\n try:\n obj = json.loads(text)\n print(f\" ✓ {text}\\n json.loads -> {obj}\")\n except Exception as e:\n print(f\" ✗ {text}\\n {e}\")"

FilesExpand file tree

Structured_Output_Guided_Decoding.ipynb

Latest commit

History

Structured_Output_Guided_Decoding.ipynb

File metadata and controls