mirror of
https://github.com/ikawrakow/ik_llama.cpp.git
synced 2026-06-28 04:30:15 -05:00
* Autoparser - complete refactoring of parser architecture Autoparser: add optional argument reshuffle capability Autoparser: True streaming (#20177) * Relax atomicity constraint for nicer, more pleasent, True Streaming parsing * Whitespace * Remove redundant atomics Revert to OAI-compatible args (#20213) * Revert to OAI-compatible args * Apply workaround::func_args_not_string Fix structured outputs (#20223) * Fix structured outputs * Update common/chat-auto-parser-generator.cpp Co-authored-by: Aldehir Rojas <hello@alde.dev> --------- Co-authored-by: Aldehir Rojas <hello@alde.dev> Fix compile bug (#20203) * Fix compile bug * Update common/chat-auto-parser-helpers.cpp Co-authored-by: Sigbjørn Skjæret <sigbjorn.skjaeret@scala.com> --------- Co-authored-by: Sigbjørn Skjæret <sigbjorn.skjaeret@scala.com> # Conflicts: # common/chat-auto-parser-helpers.cpp common : gracefully handle incomplete output (#20191) * common : handle incomplete UTF-8 at end of input in PEG parser * cont : if reached end prematurely, emit needs_more_input to propagate partial output * cont: refactor peg parse context to add lenient flag * cont : remove partial flag, keep lenient flag PEG parser for LFM2 (#20251) * PEG parser for LFM2 * Simplify using python_value() common: map developer role to system (#20215) * Map developer role to system * Simplify common: consolidate PEG string parsers (#20263) * common : consolidate PEG string parsers * cont : fix json_string_content() examples : fix empty items in json_schema_to_grammar.py [no ci] (#19968) * Fix logic for retrieving schema items in `json_schema_to_grammar.py` If `schema['items']` is `{}` and `prefixItems not in schema', as `{}` is Falsy, the original code here will raise an error. I think if `schema['items']` is `{}`, them items should just be `{}` * Apply suggestion from @CISC Co-authored-by: Sigbjørn Skjæret <sigbjorn.skjaeret@scala.com> * Add tests for arrays with empty items Add two unit tests to `tests/test-json-schema-to-grammar.cpp` that validate handling of arrays when 'items' is an empty schema and when 'prefixItems' is present alongside an empty 'items'. Both tests expect the same generated grammar, ensuring the JSON Schema->grammar conversion treats an empty 'items' schema (and the presence of 'prefixItems') correctly and covering this edge case. --------- Co-authored-by: Sigbjørn Skjæret <sigbjorn.skjaeret@scala.com> Reduce level of content parser warning message to avoid log spam on non-debug verbosity (#20347) do not return if template parse failed add arg to enable parallel tool call common : fix incorrect uses of stoul (#20313) # Conflicts: # common/arg.cpp # src/llama-grammar.cpp examples : fix empty items in json_schema_to_grammar.py [no ci] (#19968) * Fix logic for retrieving schema items in `json_schema_to_grammar.py` If `schema['items']` is `{}` and `prefixItems not in schema', as `{}` is Falsy, the original code here will raise an error. I think if `schema['items']` is `{}`, them items should just be `{}` * Apply suggestion from @CISC Co-authored-by: Sigbjørn Skjæret <sigbjorn.skjaeret@scala.com> * Add tests for arrays with empty items Add two unit tests to `tests/test-json-schema-to-grammar.cpp` that validate handling of arrays when 'items' is an empty schema and when 'prefixItems' is present alongside an empty 'items'. Both tests expect the same generated grammar, ensuring the JSON Schema->grammar conversion treats an empty 'items' schema (and the presence of 'prefixItems') correctly and covering this edge case. --------- Co-authored-by: Sigbjørn Skjæret <sigbjorn.skjaeret@scala.com> Add support for MiroThinker with new jinja template common/parser: handle reasoning budget (#20297) * v1 * Finished! * Handlie cli * Reasoning sampler * Apply suggestions from code review Co-authored-by: Sigbjørn Skjæret <sigbjorn.skjaeret@scala.com> * Less explosive terminology :) * Add utf-8 case and tests * common : migrate reasoning budget sampler to common * cont : clean up * cont : expose state and allow passing as initial state * cont : remove unused imports * cont : update state machine doc string --------- Co-authored-by: Sigbjørn Skjæret <sigbjorn.skjaeret@scala.com> Co-authored-by: Alde Rojas <hello@alde.dev> common/parser: use nlohmann::ordered_json to preserve parameter order (#20385) common/parser: add GigaChatV3/3.1 models support (#19931) Co-authored-by: Mishusha <pmv26021975@gmail.com> common/parser: gracefully handle undetected tool parser, print error message. (#20286) fix: prevent nullptr dereference (#20552) common : fix iterator::end() dereference (#20445) # Conflicts: # common/regex-partial.cpp jinja : add capability check for object args (#20612) common/parser: add `--skip-chat-parsing` to force a pure content parser. (#20289) * Add `--force-pure-content` to force a pure content parser. * Update common/arg.cpp Co-authored-by: Sigbjørn Skjæret <sigbjorn.skjaeret@scala.com> common : rework gpt-oss parser (#20393) * common : rework gpt-oss parser * cont : fix gpt-oss tests * cont : add structured output test * cont : rename final to final_msg common : fix gpt-oss content removal (#20745) common/parser: add proper reasoning tag prefill reading (#20424) * Implement proper prefill extraction * Refactor cli parameters, update docs, move reasoning budget sampler part to common/reasoning-budget.cpp * Update tools/server/server-task.cpp * refactor: move grammars to variant, remove grammar_external, handle exception internally * Make code less C++y Co-authored-by: Georgi Gerganov <ggerganov@gmail.com> chat : handle tool calls with no required args in TAG_WITH_TAGGED format (#20764) * chat : handle tool calls with no required args in TAG_WITH_TAGGED format * Update tests/test-chat.cpp [no ci] Co-authored-by: Aldehir Rojas <hello@alde.dev> --------- Co-authored-by: Piotr Wilkin (ilintar) <piotr.wilkin@syndatis.com> Co-authored-by: Aldehir Rojas <hello@alde.dev> common/parser : fix out_of_range crash in throw path (#20424 regression) (#20777) * chat : fix out_of_range crash in throw path (#20424 regression) #20424 introduced effective_input = generation_prompt + input, but the throw path uses input.substr(result.end) where result.end is a position within effective_input. Every thinking model with a non-empty generation_prompt crashes with std::out_of_range instead of the intended error message. Test crashes on unpatched master, passes with fix: cmake -B build -DLLAMA_BUILD_TESTS=ON -DLLAMA_BUILD_TOOLS=OFF cmake --build build --target test-chat ./build/bin/test-chat * Update test-chat.cpp * Update test-chat.cpp * Update test-chat.cpp --------- Co-authored-by: Piotr Wilkin (ilintar) <piotr.wilkin@syndatis.com> jinja : fix heap OOB read in value equality comparison (#20782) Address GHSA-q9j6-4hhc-rq9p and GHSA-2q4c-9gq5-5vfp. The three-iterator overload of std::equal in value_array_t::equivalent() and value_object_t::equivalent() reads past the end of the shorter container when comparing arrays or objects of different lengths. Use the four-iterator overload (C++14) which checks both range lengths. Found-by: Pwno common : fix typo in debug log ('extracft' -> 'extract') (#20807) common/parser: fix nasty bug causing subtle corruption of generation prompt (#20825) jinja : refactor token advancement (#20864) * refactor token advancement * exercise sub-expressions common/autoparser : detect reasoning markers when enable_thinking changes system prompt (#20859) common : replace wrap_for_generation with a prefix convenience function and fix gpt-oss (#20912) jinja: fix macro with kwargs (#20960) * jinja: fix macro with kwargs * Apply suggestions from code review Co-authored-by: Sigbjørn Skjæret <sigbjorn.skjaeret@scala.com> * fix newline problem --------- Co-authored-by: Sigbjørn Skjæret <sigbjorn.skjaeret@scala.com> common : inhibit lazy grammar sampler while reasoning is active (#20970) * common : inhibit grammar while reasoning budget is active * cont : update force_pos in accept * cont : fix tests * cont : tweak should apply logic * cont : return early not using grammar sampler * Add tests * cont : prevent backend sampling when reasoning budget enabled * cont : fix typo --------- Co-authored-by: Piotr Wilkin <piotr.wilkin@syndatis.com> # Conflicts: # common/reasoning-budget.h # common/sampling.cpp # tools/cli/cli.cpp # tools/server/server-common.cpp # tools/server/server-task.cpp common/parser: fix reasoning whitespace bugs + extra parser tests (#21085) * fix whitespace reasoning issues + add reconstruction tests * Proper fix * fix Nemotron autoparser test expectations to include newline in marker common : add reasoning_format = none support to gpt-oss (#21094) common/json-schema: fix: handle non-capturing groups (?:...) in JSON schema pattern converter (#21124) The regex-to-grammar converter in _visit_pattern() crashes with SIGSEGV when a JSON schema "pattern" field contains a non-capturing group (?:...). Root cause: when the parser sees '(' followed by '?', it pushes a warning but does not advance past '?:'. The recursive transform() call then interprets '?' as a quantifier and calls seq.back() on an empty vector, causing undefined behavior. This commonly occurs when serving OpenAI-compatible tool calls from clients that include complex regex patterns in their JSON schemas (e.g., date validation patterns like ^(?:(?:\d\d[2468][048]|...)-02-29|...)$). The fix: - Skip '?:' after '(' to treat non-capturing groups as regular groups - For unsupported syntax (?=, ?!, etc.), skip to matching ')' safely, handling escaped characters to avoid miscounting parenthesis depth - Adjust the ')' unbalanced-parentheses check using direct char comparisons instead of substr - Add test cases for non-capturing groups (C++ only, as the JS/Python implementations do not yet support this syntax) common/parser: fix handling of tool definition with missing properties key (#21128) jinja : handle empty expressions correctly (#20913) * Reject empty computed member expressions before returning slices[0] from parse_member_expression_arguments(). * Treat empty computed member expressions with Jinja2 undefined semantics Treat empty computed member expressions like `a[]` as undefined instead of raising a parser error, to match Jinja2 behavior. - return a noop expression for empty computed member arguments - return undefined when a computed member key evaluates to undefined - add Jinja tests covering `a[]|default('fallback')` and `a[] is undefined` * Handle undefined computed member properties Move undefined-property handling to the common member access path, and add a test covering `a[undefined] is undefined`. * Use default undefined value in member access Initialize val and then return it when property is undefined. Co-authored-by: Sigbjørn Skjæret <sigbjorn.skjaeret@scala.com> * empty statement parses to blank_expression instead of noop_statement --------- Co-authored-by: Sigbjørn Skjæret <sigbjorn.skjaeret@scala.com> common : gpt-oss handle builtin and unsolicited tool calls (#21213) fix: tool call parsing for LFM2 and LFM2.5 models (#21242) * fix: tool call parsing for LFM2 and LFM2.5 models' * refactor: add test / break out lfm2 and lfm2.5 parsing logic # Conflicts: # common/chat.cpp Relax prefill parser to allow space. (#21240) * Relax prefill parser to allow space. * Move changes from prefix() to parser generation * Only allow spaces if we're not having a pure content parser next common : add commentary rules for gpt-oss-20b (#21286) add reasoning budget model, mtmd: fix gguf conversion for audio/vision mmproj (#21309) * fix gguf conversion for audio/vision mmproj * fix test # Conflicts: # convert_hf_to_gguf.py # examples/eval-callback/eval-callback.cpp # examples/mtmd/CMakeLists.txt # examples/mtmd/clip-impl.h # examples/mtmd/mtmd.cpp # gguf-py/gguf/constants.py # gguf-py/gguf/gguf_writer.py # gguf-py/gguf/tensor_mapping.py # src/CMakeLists.txt # src/llama-arch.cpp # src/llama-arch.h # src/llama-model.cpp # src/llama-model.h # src/llama-vocab.cpp # src/models/models.h # tests/test-llama-archs.cpp # tools/mtmd/clip-graph.h # tools/mtmd/clip-model.h # tools/mtmd/clip.cpp # tools/mtmd/models/models.h fix: gemma 4 template (#21326) chat : avoid including json in chat.h (#21306) jinja: coerce input for string-specific filters (#21370) common : fix tool call type detection for nullable and enum schemas (#21327) * common : fix tool call type detection for nullable and enum schemas * common, tests : fix grammar delegation for nullable/enum schemas and add tests Fix enum type inference to scan all enum values (not just index 0) so schemas like {"enum": [0, "celsius"]} correctly detect string type. Fix schema_delegates in peg-parser to handle nullable type arrays (["string", "null"]) and typeless enum schemas in raw mode, allowing the tagged parser to use raw text instead of JSON-formatted strings. Add test cases for Qwen3-Coder (TAG_WITH_TAGGED format): - nullable string ["string", "null"] - nullable string with null first ["null", "string"] - nullable integer ["integer", "null"] - enum without explicit type key common/parser: fix call ID detection (Mistral parser mostly) + atomicity for tag-json parsers (#21230) * Fix call ID detection (Mistral parser mostly) + atomicity for tag-json parsers * Rename * Update common/chat-auto-parser-generator.cpp Co-authored-by: Sigbjørn Skjæret <sigbjorn.skjaeret@scala.com> --------- Co-authored-by: Sigbjørn Skjæret <sigbjorn.skjaeret@scala.com> common : add gemma 4 specialized parser (#21418) * common : add gemma4 dedicated parser * cont : add '<|tool_response>' as eog * cont : emit JSON from Gemma4 tool call AST * cont : more fixes * cont : refactor convert function * cont : refine rules and mapping * cont : add more tests * cont : clean up * cont : remove autoparser gemma4 implementation * cont : more cleanup * cont : rename gemma4.jinja to match the others * cont : add custom template to support interleaved thinking * cont : preserve reasoning in model turns * cont : fix initializer error * cont : fix unused vars * cont : fix accidental static * cont : fix specialized_template signature * fix extra semicolon * remove debug line and extra space [no ci] fix reasoning budget parser: fix MiniMax handling (#21573) jinja : support ensure_ascii=true, string repetition and int/float self-filtering (#21623) * feat: jinja engine improvements for reka-edge Port three Jinja engine improvements needed for the reka-edge model: 1. Python-style string repetition ("ab" * 3 → "ababab") 2. ensure_ascii=true support for tojson filter (escapes non-ASCII to \uXXXX) 3. int() builtin on value_int_t (identity, needed for Reka Edge template) * fix: escape invalid utf8 bytes when ensure_ascii=true The json_ensure_ascii_preserving_format function does not correctly handle an edge case where if UTF-8 parsing fails, it adds the non-ascii character back to the output as a raw byte. This commit fixes that by adding the unicode standard replacement character \\ufffd to the output instead. This is the standard behavior for various programming languages like Python, Rust, Go, etc. * chore: address PR comments 1. Add todo comment for supporting string repetition for array/tuples 2. Add support for float identity operation 3. Move invalid ascii test case to test_fuzzing * chore: accept suggestion for common/jinja/value.cpp Co-authored-by: Sigbjørn Skjæret <sigbjorn.skjaeret@scala.com> --------- Co-authored-by: Sigbjørn Skjæret <sigbjorn.skjaeret@scala.com> common : simplify autoparser tagged parser rules (#21216) * common : simplify autoparser tagged parser rules * cont : remove upper limit on optional args * cont : revert changes to parsing at the end * cont : undo arbitrary ordering of optional args * cont : fix uninitialized required parameters * revert to simplify merge * re-apply patches * restore flexible optional arg ordering tests common : fix ambiguous grammar rule in gemma4 (#21661) * common : fix ambiguous grammar rule in gemma4 * cont : fix missing comma... common : enable reasoning budget sampler for gemma4 (#21697) * fix: enable reasoning budget sampler for gemma4 Add thinking_start_tag and thinking_end_tag to common_chat_params_init_gemma4(). Without these, the reasoning budget sampler never activates for gemma4. Make the newline after "thought" optional in the PEG parser to handle budget=0 (sampler forces end tag before the newline). Add test case for empty thinking block. Fixes #21487 * use p.space() instead of p.optional(p.literal("\n")) in gemma4 thought parser common : better align to the updated official gemma4 template (#21704) fix: Fix broken structured output when using $refs in json_schema (#21699) chat: dedicated DeepSeek v3.2 parser + "official" template (#21785) Hide render_message_to_json warning common/gemma4 : handle parsing edge cases (#21760) common: skip reasoning budget sampler when no budget is requested (#21870) * common: skip reasoning budget sampler when no budget is requested After I added thinking_start_tag / thinking_end_tag for gemma4 in #21697, the reasoning budget sampler gets unconditionally created even when no budget is configured (the default -1). The same applies to kimi_k2, lfm2, lfm2_5, and ministral_3 which also set these tags. The budget gets converted to INT_MAX, so the sampler never actually forces any tokens but still runs per-token checks (start tag matching in IDLE state, token-to-piece conversion + UTF-8 checks in COUNTING state). More importantly, the mere existence of the sampler (non-null rbudget) disables backend sampling. Backend sampling lets the GPU select tokens directly, avoiding a full logits transfer from GPU to CPU every token. This could explain the 30% speed regression reported in #21784 (98 t/s to 70 t/s on Vulkan). So I added a reasoning_budget_tokens >= 0 check to the sampler creation condition. When the budget is unlimited, the sampler is not created, backend sampling stays enabled, and no per-token overhead is added. When a budget is explicitly set (0, 128, 1024, etc.), the sampler is created and works as before. * common: preserve rbudget when grammar is lazy Following up on the review feedback on #21870: keep the reasoning budget sampler when grammar_lazy is true, so the thinking-block grammar suppression from #20970 still works when tools are in use. This way, we only skip the sampler when both no budget is set AND grammar is not lazy. autoparser: support case of JSON_NATIVE with per-call markers (test case: Reka-Edge) (#21892) * fix grammar * fix add sampled token --------- Co-authored-by: Piotr Wilkin (ilintar) <piotr.wilkin@syndatis.com> Co-authored-by: firecoperana <firecoperana>
843 lines
34 KiB
Python
Executable File
843 lines
34 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import itertools
|
|
import json
|
|
import re
|
|
import sys
|
|
from typing import Any, List, Optional, Set, Tuple, Union
|
|
|
|
def _build_repetition(item_rule, min_items, max_items, separator_rule=None):
|
|
|
|
if max_items == 0:
|
|
return ""
|
|
|
|
if min_items == 0 and max_items == 1:
|
|
return f'{item_rule}?'
|
|
|
|
if not separator_rule:
|
|
if min_items == 1 and max_items is None:
|
|
return f'{item_rule}+'
|
|
elif min_items == 0 and max_items is None:
|
|
return f'{item_rule}*'
|
|
else:
|
|
return f'{item_rule}{{{min_items},{max_items if max_items is not None else ""}}}'
|
|
|
|
result = item_rule + ' ' + _build_repetition(f'({separator_rule} {item_rule})', min_items - 1 if min_items > 0 else 0, max_items - 1 if max_items is not None else None)
|
|
return f'({result})?' if min_items == 0 else result
|
|
|
|
def _generate_min_max_int(min_value: Optional[int], max_value: Optional[int], out: list, decimals_left: int = 16, top_level: bool = True):
|
|
has_min = min_value != None
|
|
has_max = max_value != None
|
|
|
|
def digit_range(from_char: str, to_char: str):
|
|
out.append("[")
|
|
if from_char == to_char:
|
|
out.append(from_char)
|
|
else:
|
|
out.append(from_char)
|
|
out.append("-")
|
|
out.append(to_char)
|
|
out.append("]")
|
|
|
|
def more_digits(min_digits: int, max_digits: int):
|
|
out.append("[0-9]")
|
|
if min_digits == max_digits and min_digits == 1:
|
|
return
|
|
out.append("{")
|
|
out.append(str(min_digits))
|
|
if max_digits != min_digits:
|
|
out.append(",")
|
|
if max_digits != sys.maxsize:
|
|
out.append(str(max_digits))
|
|
out.append("}")
|
|
|
|
def uniform_range(from_str: str, to_str: str):
|
|
i = 0
|
|
while i < len(from_str) and from_str[i] == to_str[i]:
|
|
i += 1
|
|
if i > 0:
|
|
out.append("\"")
|
|
out.append(from_str[:i])
|
|
out.append("\"")
|
|
if i < len(from_str):
|
|
if i > 0:
|
|
out.append(" ")
|
|
sub_len = len(from_str) - i - 1
|
|
if sub_len > 0:
|
|
from_sub = from_str[i+1:]
|
|
to_sub = to_str[i+1:]
|
|
sub_zeros = "0" * sub_len
|
|
sub_nines = "9" * sub_len
|
|
|
|
to_reached = False
|
|
out.append("(")
|
|
if from_sub == sub_zeros:
|
|
digit_range(from_str[i], chr(ord(to_str[i]) - 1))
|
|
out.append(" ")
|
|
more_digits(sub_len, sub_len)
|
|
else:
|
|
out.append("[")
|
|
out.append(from_str[i])
|
|
out.append("] ")
|
|
out.append("(")
|
|
uniform_range(from_sub, sub_nines)
|
|
out.append(")")
|
|
if ord(from_str[i]) < ord(to_str[i]) - 1:
|
|
out.append(" | ")
|
|
if to_sub == sub_nines:
|
|
digit_range(chr(ord(from_str[i]) + 1), to_str[i])
|
|
to_reached = True
|
|
else:
|
|
digit_range(chr(ord(from_str[i]) + 1), chr(ord(to_str[i]) - 1))
|
|
out.append(" ")
|
|
more_digits(sub_len, sub_len)
|
|
if not to_reached:
|
|
out.append(" | ")
|
|
digit_range(to_str[i], to_str[i])
|
|
out.append(" ")
|
|
uniform_range(sub_zeros, to_sub)
|
|
out.append(")")
|
|
else:
|
|
out.append("[")
|
|
out.append(from_str[i])
|
|
out.append("-")
|
|
out.append(to_str[i])
|
|
out.append("]")
|
|
|
|
if has_min and has_max:
|
|
if min_value < 0 and max_value < 0:
|
|
out.append("\"-\" (")
|
|
_generate_min_max_int(-max_value, -min_value, out, decimals_left, top_level=True)
|
|
out.append(")")
|
|
return
|
|
|
|
if min_value < 0:
|
|
out.append("\"-\" (")
|
|
_generate_min_max_int(0, -min_value, out, decimals_left, top_level=True)
|
|
out.append(") | ")
|
|
min_value = 0
|
|
|
|
min_s = str(min_value)
|
|
max_s = str(max_value)
|
|
min_digits = len(min_s)
|
|
max_digits = len(max_s)
|
|
|
|
for digits in range(min_digits, max_digits):
|
|
uniform_range(min_s, "9" * digits)
|
|
min_s = "1" + "0" * digits
|
|
out.append(" | ")
|
|
uniform_range(min_s, max_s)
|
|
return
|
|
|
|
less_decimals = max(decimals_left - 1, 1)
|
|
|
|
if has_min:
|
|
if min_value < 0:
|
|
out.append("\"-\" (")
|
|
_generate_min_max_int(None, -min_value, out, decimals_left, top_level=False)
|
|
out.append(") | [0] | [1-9] ")
|
|
more_digits(0, decimals_left - 1)
|
|
elif min_value == 0:
|
|
if top_level:
|
|
out.append("[0] | [1-9] ")
|
|
more_digits(0, less_decimals)
|
|
else:
|
|
more_digits(1, decimals_left)
|
|
elif min_value <= 9:
|
|
c = str(min_value)
|
|
range_start = '1' if top_level else '0'
|
|
if c > range_start:
|
|
digit_range(range_start, chr(ord(c) - 1))
|
|
out.append(" ")
|
|
more_digits(1, less_decimals)
|
|
out.append(" | ")
|
|
digit_range(c, "9")
|
|
out.append(" ")
|
|
more_digits(0, less_decimals)
|
|
else:
|
|
min_s = str(min_value)
|
|
length = len(min_s)
|
|
c = min_s[0]
|
|
|
|
if c > "1":
|
|
digit_range("1" if top_level else "0", chr(ord(c) - 1))
|
|
out.append(" ")
|
|
more_digits(length, less_decimals)
|
|
out.append(" | ")
|
|
digit_range(c, c)
|
|
out.append(" (")
|
|
_generate_min_max_int(int(min_s[1:]), None, out, less_decimals, top_level=False)
|
|
out.append(")")
|
|
if c < "9":
|
|
out.append(" | ")
|
|
digit_range(chr(ord(c) + 1), "9")
|
|
out.append(" ")
|
|
more_digits(length - 1, less_decimals)
|
|
return
|
|
|
|
if has_max:
|
|
if max_value >= 0:
|
|
if top_level:
|
|
out.append("\"-\" [1-9] ")
|
|
more_digits(0, less_decimals)
|
|
out.append(" | ")
|
|
_generate_min_max_int(0, max_value, out, decimals_left, top_level=True)
|
|
else:
|
|
out.append("\"-\" (")
|
|
_generate_min_max_int(-max_value, None, out, decimals_left, top_level=False)
|
|
out.append(")")
|
|
return
|
|
|
|
raise RuntimeError("At least one of min_value or max_value must be set")
|
|
|
|
class BuiltinRule:
|
|
def __init__(self, content: str, deps: list | None = None):
|
|
self.content = content
|
|
self.deps = deps or []
|
|
|
|
# Constraining spaces to prevent model "running away".
|
|
SPACE_RULE = '| " " | "\\n"{1,2} [ \\t]{0,20}'
|
|
|
|
PRIMITIVE_RULES = {
|
|
'boolean' : BuiltinRule('("true" | "false") space', []),
|
|
'decimal-part' : BuiltinRule('[0-9]{1,16}', []),
|
|
'integral-part': BuiltinRule('[0] | [1-9] [0-9]{0,15}', []),
|
|
'number' : BuiltinRule('("-"? integral-part) ("." decimal-part)? ([eE] [-+]? integral-part)? space', ['integral-part', 'decimal-part']),
|
|
'integer' : BuiltinRule('("-"? integral-part) space', ['integral-part']),
|
|
'value' : BuiltinRule('object | array | string | number | boolean | null', ['object', 'array', 'string', 'number', 'boolean', 'null']),
|
|
'object' : BuiltinRule('"{" space ( string ":" space value ("," space string ":" space value)* )? "}" space', ['string', 'value']),
|
|
'array' : BuiltinRule('"[" space ( value ("," space value)* )? "]" space', ['value']),
|
|
'uuid' : BuiltinRule(r'"\"" [0-9a-fA-F]{8} "-" [0-9a-fA-F]{4} "-" [0-9a-fA-F]{4} "-" [0-9a-fA-F]{4} "-" [0-9a-fA-F]{12} "\"" space', []),
|
|
'char' : BuiltinRule(r'[^"\\\x7F\x00-\x1F] | [\\] (["\\bfnrt] | "u" [0-9a-fA-F]{4})', []),
|
|
'string' : BuiltinRule(r'"\"" char* "\"" space', ['char']),
|
|
'null' : BuiltinRule('"null" space', []),
|
|
}
|
|
|
|
# TODO: support "uri", "email" string formats
|
|
STRING_FORMAT_RULES = {
|
|
'date' : BuiltinRule('[0-9]{4} "-" ( "0" [1-9] | "1" [0-2] ) "-" ( \"0\" [1-9] | [1-2] [0-9] | "3" [0-1] )', []),
|
|
'time' : BuiltinRule('([01] [0-9] | "2" [0-3]) ":" [0-5] [0-9] ":" [0-5] [0-9] ( "." [0-9]{3} )? ( "Z" | ( "+" | "-" ) ( [01] [0-9] | "2" [0-3] ) ":" [0-5] [0-9] )', []),
|
|
'date-time' : BuiltinRule('date "T" time', ['date', 'time']),
|
|
'date-string' : BuiltinRule('"\\"" date "\\"" space', ['date']),
|
|
'time-string' : BuiltinRule('"\\"" time "\\"" space', ['time']),
|
|
'date-time-string': BuiltinRule('"\\"" date-time "\\"" space', ['date-time']),
|
|
}
|
|
|
|
DOTALL = '[\\U00000000-\\U0010FFFF]'
|
|
DOT = '[^\\x0A\\x0D]'
|
|
|
|
RESERVED_NAMES = set(["root", "dot", *PRIMITIVE_RULES.keys(), *STRING_FORMAT_RULES.keys()])
|
|
|
|
INVALID_RULE_CHARS_RE = re.compile(r'[^a-zA-Z0-9-]+')
|
|
GRAMMAR_LITERAL_ESCAPE_RE = re.compile(r'[\r\n"\\]')
|
|
GRAMMAR_RANGE_LITERAL_ESCAPE_RE = re.compile(r'[\r\n"\]\-\\]')
|
|
GRAMMAR_LITERAL_ESCAPES = {'\r': '\\r', '\n': '\\n', '"': '\\"', '-': '\\-', ']': '\\]', '\\': '\\\\'}
|
|
|
|
NON_LITERAL_SET = set('|.()[]{}*+?')
|
|
ESCAPED_IN_REGEXPS_BUT_NOT_IN_LITERALS = set('^$.[]()|{}*+?')
|
|
|
|
|
|
class SchemaConverter:
|
|
def __init__(self, *, prop_order, allow_fetch, dotall, raw_pattern):
|
|
self._prop_order = prop_order
|
|
self._allow_fetch = allow_fetch
|
|
self._dotall = dotall
|
|
self._raw_pattern = raw_pattern
|
|
self._rules = {
|
|
'space': SPACE_RULE,
|
|
}
|
|
self._refs = {}
|
|
self._refs_being_resolved = set()
|
|
|
|
def _format_literal(self, literal):
|
|
escaped = GRAMMAR_LITERAL_ESCAPE_RE.sub(
|
|
lambda m: GRAMMAR_LITERAL_ESCAPES.get(m.group(0)) or m.group(0), literal
|
|
)
|
|
return f'"{escaped}"'
|
|
|
|
def not_literal(self, literal: str, dotall: bool = True, maybe_escaped_underscores = False) -> str:
|
|
'''
|
|
not_literal('a') -> '[^a]'
|
|
not_literal('abc') -> '([^a] | "a" ([^b] | "b" ([^c])?)?)?'
|
|
'''
|
|
assert len(literal) > 0, 'Empty literal not supported'
|
|
def recurse(i: int):
|
|
c = literal[i]
|
|
if maybe_escaped_underscores and c == '_':
|
|
yield f'[^{c}\\\\]'
|
|
yield ' | '
|
|
yield f'"\\\\"? "{c}"'
|
|
else:
|
|
yield f'[^{c}]'
|
|
if i < len(literal) - 1:
|
|
yield ' | '
|
|
yield self._format_literal(c)
|
|
yield ' ('
|
|
yield from recurse(i + 1)
|
|
yield ')?'
|
|
|
|
return ''.join(('(', *recurse(0), ')'))
|
|
|
|
def _not_strings(self, strings):
|
|
class TrieNode:
|
|
def __init__(self):
|
|
self.children = {}
|
|
self.is_end_of_string = False
|
|
|
|
def insert(self, string):
|
|
node = self
|
|
for c in string:
|
|
node = node.children.setdefault(c, TrieNode())
|
|
node.is_end_of_string = True
|
|
|
|
trie = TrieNode()
|
|
for s in strings:
|
|
trie.insert(s)
|
|
|
|
char_rule = self._add_primitive('char', PRIMITIVE_RULES['char'])
|
|
out = ['["] ( ']
|
|
|
|
def visit(node):
|
|
rejects = []
|
|
first = True
|
|
for c in sorted(node.children.keys()):
|
|
child = node.children[c]
|
|
rejects.append(c)
|
|
if first:
|
|
first = False
|
|
else:
|
|
out.append(' | ')
|
|
out.append(f'[{c}]')
|
|
if child.children:
|
|
out.append(f' (')
|
|
visit(child)
|
|
out.append(')')
|
|
elif child.is_end_of_string:
|
|
out.append(f' {char_rule}+')
|
|
if node.children:
|
|
if not first:
|
|
out.append(' | ')
|
|
out.append(f'[^"{"".join(rejects)}] {char_rule}*')
|
|
visit(trie)
|
|
|
|
out.append(f' ){"" if trie.is_end_of_string else "?"} ["] space')
|
|
return ''.join(out)
|
|
|
|
def _add_rule(self, name, rule):
|
|
esc_name = INVALID_RULE_CHARS_RE.sub('-', name)
|
|
if esc_name not in self._rules or self._rules[esc_name] == rule:
|
|
key = esc_name
|
|
else:
|
|
i = 0
|
|
while f'{esc_name}{i}' in self._rules and self._rules[f'{esc_name}{i}'] != rule:
|
|
i += 1
|
|
key = f'{esc_name}{i}'
|
|
self._rules[key] = rule
|
|
return key
|
|
|
|
def resolve_refs(self, schema: dict, url: str):
|
|
'''
|
|
Resolves all $ref fields in the given schema, fetching any remote schemas,
|
|
replacing $ref with absolute reference URL and populating self._refs with the
|
|
respective referenced (sub)schema dictionaries.
|
|
'''
|
|
def visit(n: dict):
|
|
if isinstance(n, list):
|
|
return [visit(x) for x in n]
|
|
elif isinstance(n, dict):
|
|
ref = n.get('$ref')
|
|
if ref is not None and ref not in self._refs:
|
|
if ref.startswith('https://'):
|
|
assert self._allow_fetch, 'Fetching remote schemas is not allowed (use --allow-fetch for force)'
|
|
import requests
|
|
|
|
frag_split = ref.split('#')
|
|
base_url = frag_split[0]
|
|
|
|
target = self._refs.get(base_url)
|
|
if target is None:
|
|
target = self.resolve_refs(requests.get(ref).json(), base_url)
|
|
self._refs[base_url] = target
|
|
|
|
if len(frag_split) == 1 or frag_split[-1] == '':
|
|
return target
|
|
elif ref.startswith('#/'):
|
|
target = schema
|
|
ref = f'{url}{ref}'
|
|
n['$ref'] = ref
|
|
else:
|
|
raise ValueError(f'Unsupported ref {ref}')
|
|
|
|
for sel in ref.split('#')[-1].split('/')[1:]:
|
|
assert target is not None, f'Error resolving ref {ref}: {sel} not in {target}'
|
|
if isinstance(target, list):
|
|
try:
|
|
sel_index = int(sel)
|
|
except ValueError:
|
|
raise ValueError(f'Error resolving ref {ref}: {sel} not in {target}')
|
|
assert 0 <= sel_index < len(target), f'Error resolving ref {ref}: {sel} not in {target}'
|
|
target = target[sel_index]
|
|
else:
|
|
assert sel in target, f'Error resolving ref {ref}: {sel} not in {target}'
|
|
target = target[sel]
|
|
|
|
self._refs[ref] = target
|
|
else:
|
|
for v in n.values():
|
|
visit(v)
|
|
|
|
return n
|
|
return visit(schema)
|
|
|
|
def _generate_union_rule(self, name, alt_schemas):
|
|
return ' | '.join((
|
|
self.visit(alt_schema, f'{name}{"-" if name else "alternative-"}{i}')
|
|
for i, alt_schema in enumerate(alt_schemas)
|
|
))
|
|
|
|
def _visit_pattern(self, pattern, name):
|
|
'''
|
|
Transforms a regular expression pattern into a GBNF rule.
|
|
|
|
Input: https://json-schema.org/understanding-json-schema/reference/regular_expressions
|
|
Output: https://github.com/ggerganov/llama.cpp/blob/master/grammars/README.md
|
|
|
|
Unsupported features: negative/positive lookaheads, greedy/non-greedy modifiers.
|
|
|
|
Mostly a 1:1 translation, except for {x} / {x,} / {x,y} quantifiers for which
|
|
we define sub-rules to keep the output lean.
|
|
'''
|
|
|
|
assert pattern.startswith('^') and pattern.endswith('$'), 'Pattern must start with "^" and end with "$"'
|
|
pattern = pattern[1:-1]
|
|
sub_rule_ids = {}
|
|
|
|
i = 0
|
|
length = len(pattern)
|
|
|
|
def to_rule(s: tuple[str, bool]) -> str:
|
|
(txt, is_literal) = s
|
|
return "\"" + txt + "\"" if is_literal else txt
|
|
|
|
def transform() -> tuple[str, bool]:
|
|
'''
|
|
Parse a unit at index i (advancing it), and return its string representation + whether it's a literal.
|
|
'''
|
|
nonlocal i
|
|
nonlocal pattern
|
|
nonlocal sub_rule_ids
|
|
|
|
start = i
|
|
# For each component of this sequence, store its string representation and whether it's a literal.
|
|
# We only need a flat structure here to apply repetition operators to the last item, and
|
|
# to merge literals at the and (we're parsing grouped ( sequences ) recursively and don't treat '|' specially
|
|
# (GBNF's syntax is luckily very close to regular expressions!)
|
|
seq: list[tuple[str, bool]] = []
|
|
|
|
def get_dot():
|
|
if self._dotall:
|
|
rule = DOTALL
|
|
else:
|
|
# Accept any character... except \n and \r line break chars (\x0A and \xOD)
|
|
rule = DOT
|
|
return self._add_rule(f'dot', rule)
|
|
|
|
def join_seq():
|
|
nonlocal seq
|
|
ret = []
|
|
for is_literal, g in itertools.groupby(seq, lambda x: x[1]):
|
|
if is_literal:
|
|
ret.append((''.join(x[0] for x in g), True))
|
|
else:
|
|
ret.extend(g)
|
|
if len(ret) == 1:
|
|
return ret[0]
|
|
return (' '.join(to_rule(x) for x in seq), False)
|
|
|
|
while i < length:
|
|
c = pattern[i]
|
|
if c == '.':
|
|
seq.append((get_dot(), False))
|
|
i += 1
|
|
elif c == '(':
|
|
i += 1
|
|
if i < length:
|
|
assert pattern[i] != '?', f'Unsupported pattern syntax "{pattern[i]}" at index {i} of /{pattern}/'
|
|
seq.append((f'({to_rule(transform())})', False))
|
|
elif c == ')':
|
|
i += 1
|
|
assert start > 0 and pattern[start-1] == '(', f'Unbalanced parentheses; start = {start}, i = {i}, pattern = {pattern}'
|
|
return join_seq()
|
|
elif c == '[':
|
|
square_brackets = c
|
|
i += 1
|
|
while i < length and pattern[i] != ']':
|
|
if pattern[i] == '\\':
|
|
square_brackets += pattern[i:i+2]
|
|
i += 2
|
|
else:
|
|
square_brackets += pattern[i]
|
|
i += 1
|
|
assert i < length, f'Unbalanced square brackets; start = {start}, i = {i}, pattern = {pattern}'
|
|
square_brackets += ']'
|
|
i += 1
|
|
seq.append((square_brackets, False))
|
|
elif c == '|':
|
|
seq.append(('|', False))
|
|
i += 1
|
|
elif c in ('*', '+', '?'):
|
|
seq[-1] = (to_rule(seq[-1]) + c, False)
|
|
i += 1
|
|
elif c == '{':
|
|
curly_brackets = c
|
|
i += 1
|
|
while i < length and pattern[i] != '}':
|
|
curly_brackets += pattern[i]
|
|
i += 1
|
|
assert i < length, f'Unbalanced curly brackets; start = {start}, i = {i}, pattern = {pattern}'
|
|
curly_brackets += '}'
|
|
i += 1
|
|
nums = [s.strip() for s in curly_brackets[1:-1].split(',')]
|
|
min_times = 0
|
|
max_times = None
|
|
try:
|
|
if len(nums) == 1:
|
|
min_times = int(nums[0])
|
|
max_times = min_times
|
|
else:
|
|
assert len(nums) == 2
|
|
min_times = int(nums[0]) if nums[0] else 0
|
|
max_times = int(nums[1]) if nums[1] else None
|
|
except ValueError:
|
|
raise ValueError(f'Invalid quantifier {curly_brackets} in /{pattern}/')
|
|
|
|
(sub, sub_is_literal) = seq[-1]
|
|
|
|
if not sub_is_literal:
|
|
id = sub_rule_ids.get(sub)
|
|
if id is None:
|
|
id = self._add_rule(f'{name}-{len(sub_rule_ids) + 1}', sub)
|
|
sub_rule_ids[sub] = id
|
|
sub = id
|
|
|
|
seq[-1] = (_build_repetition(f'"{sub}"' if sub_is_literal else sub, min_times, max_times), False)
|
|
else:
|
|
literal = ''
|
|
while i < length:
|
|
if pattern[i] == '\\' and i < length - 1:
|
|
next = pattern[i + 1]
|
|
if next in ESCAPED_IN_REGEXPS_BUT_NOT_IN_LITERALS:
|
|
i += 1
|
|
literal += pattern[i]
|
|
i += 1
|
|
else:
|
|
literal += pattern[i:i+2]
|
|
i += 2
|
|
elif pattern[i] == '"' and not self._raw_pattern:
|
|
literal += '\\"'
|
|
i += 1
|
|
elif pattern[i] not in NON_LITERAL_SET and \
|
|
(i == length - 1 or literal == '' or pattern[i+1] == '.' or pattern[i+1] not in NON_LITERAL_SET):
|
|
literal += pattern[i]
|
|
i += 1
|
|
else:
|
|
break
|
|
if literal:
|
|
seq.append((literal, True))
|
|
|
|
return join_seq()
|
|
|
|
return self._add_rule(
|
|
name,
|
|
to_rule(transform()) if self._raw_pattern \
|
|
else "\"\\\"\" (" + to_rule(transform()) + ") \"\\\"\" space")
|
|
|
|
|
|
def _resolve_ref(self, ref):
|
|
ref_fragment = ref.split('#')[-1]
|
|
ref_name = 'ref' + re.sub(r'[^a-zA-Z0-9-]+', '-', ref_fragment)
|
|
if ref_name not in self._rules and ref not in self._refs_being_resolved:
|
|
self._refs_being_resolved.add(ref)
|
|
resolved = self._refs[ref]
|
|
ref_name = self.visit(resolved, ref_name)
|
|
self._refs_being_resolved.remove(ref)
|
|
return ref_name
|
|
|
|
def _generate_constant_rule(self, value):
|
|
return self._format_literal(json.dumps(value))
|
|
|
|
def visit(self, schema, name):
|
|
schema_type = schema.get('type')
|
|
schema_format = schema.get('format')
|
|
rule_name = name + '-' if name in RESERVED_NAMES else name or 'root'
|
|
|
|
if (ref := schema.get('$ref')) is not None:
|
|
return self._add_rule(rule_name, self._resolve_ref(ref))
|
|
|
|
elif 'oneOf' in schema or 'anyOf' in schema:
|
|
return self._add_rule(rule_name, self._generate_union_rule(name, schema.get('oneOf') or schema['anyOf']))
|
|
|
|
elif isinstance(schema_type, list):
|
|
return self._add_rule(rule_name, self._generate_union_rule(name, [{**schema, 'type': t} for t in schema_type]))
|
|
|
|
elif 'const' in schema:
|
|
return self._add_rule(rule_name, self._generate_constant_rule(schema['const']) + ' space')
|
|
|
|
elif 'enum' in schema:
|
|
rule = '(' + ' | '.join((self._generate_constant_rule(v) for v in schema['enum'])) + ') space'
|
|
return self._add_rule(rule_name, rule)
|
|
|
|
elif schema_type in (None, 'object') and \
|
|
('properties' in schema or \
|
|
('additionalProperties' in schema and schema['additionalProperties'] is not True)):
|
|
required = set(schema.get('required', []))
|
|
properties = list(schema.get('properties', {}).items())
|
|
return self._add_rule(rule_name, self._build_object_rule(properties, required, name, schema.get('additionalProperties')))
|
|
|
|
elif schema_type in (None, 'object', 'string') and 'allOf' in schema:
|
|
required = set()
|
|
properties = []
|
|
enum_sets = []
|
|
hybrid_name = name
|
|
def add_component(comp_schema, is_required):
|
|
if (ref := comp_schema.get('$ref')) is not None:
|
|
comp_schema = self._refs[ref]
|
|
|
|
if 'properties' in comp_schema:
|
|
for prop_name, prop_schema in comp_schema['properties'].items():
|
|
properties.append((prop_name, prop_schema))
|
|
if is_required:
|
|
required.add(prop_name)
|
|
|
|
if 'enum' in comp_schema:
|
|
enum_sets.append(set(comp_schema['enum']))
|
|
|
|
for t in schema['allOf']:
|
|
if 'anyOf' in t:
|
|
for tt in t['anyOf']:
|
|
add_component(tt, is_required=False)
|
|
else:
|
|
add_component(t, is_required=True)
|
|
|
|
if enum_sets:
|
|
enum_intersection = enum_sets[0]
|
|
for s in enum_sets[1:]:
|
|
enum_intersection &= s
|
|
|
|
if enum_intersection:
|
|
rule = '(' + ' | '.join((self._generate_constant_rule(v) for v in sorted(enum_intersection))) + ') space'
|
|
return self._add_rule(rule_name, rule)
|
|
|
|
return self._add_rule(rule_name, self._build_object_rule(properties, required, hybrid_name, additional_properties=None))
|
|
|
|
elif schema_type in (None, 'array') and ('items' in schema or 'prefixItems' in schema):
|
|
items = schema.get('items', schema.get('prefixItems'))
|
|
if isinstance(items, list):
|
|
return self._add_rule(
|
|
rule_name,
|
|
'"[" space ' +
|
|
' "," space '.join(
|
|
self.visit(item, f'{name}{"-" if name else ""}tuple-{i}')
|
|
for i, item in enumerate(items)) +
|
|
' "]" space')
|
|
else:
|
|
item_rule_name = self.visit(items, f'{name}{"-" if name else ""}item')
|
|
min_items = schema.get("minItems", 0)
|
|
max_items = schema.get("maxItems")
|
|
return self._add_rule(rule_name, '"[" space ' + _build_repetition(item_rule_name, min_items, max_items, separator_rule='"," space') + ' "]" space')
|
|
|
|
elif schema_type in (None, 'string') and 'pattern' in schema:
|
|
return self._visit_pattern(schema['pattern'], rule_name)
|
|
|
|
elif schema_type in (None, 'string') and re.match(r'^uuid[1-5]?$', schema_format or ''):
|
|
return self._add_primitive(
|
|
'root' if rule_name == 'root' else schema_format,
|
|
PRIMITIVE_RULES['uuid']
|
|
)
|
|
|
|
elif schema_type in (None, 'string') and f'{schema_format}-string' in STRING_FORMAT_RULES:
|
|
prim_name = f'{schema_format}-string'
|
|
return self._add_rule(rule_name, self._add_primitive(prim_name, STRING_FORMAT_RULES[prim_name]))
|
|
|
|
elif schema_type == 'string' and ('minLength' in schema or 'maxLength' in schema):
|
|
char_rule = self._add_primitive('char', PRIMITIVE_RULES['char'])
|
|
min_len = schema.get('minLength', 0)
|
|
max_len = schema.get('maxLength')
|
|
|
|
return self._add_rule(rule_name, r'"\"" ' + _build_repetition(char_rule, min_len, max_len) + r' "\"" space')
|
|
|
|
elif schema_type in (None, 'integer') and \
|
|
('minimum' in schema or 'exclusiveMinimum' in schema or 'maximum' in schema or 'exclusiveMaximum' in schema):
|
|
min_value = None
|
|
max_value = None
|
|
if 'minimum' in schema:
|
|
min_value = schema['minimum']
|
|
elif 'exclusiveMinimum' in schema:
|
|
min_value = schema['exclusiveMinimum'] + 1
|
|
if 'maximum' in schema:
|
|
max_value = schema['maximum']
|
|
elif 'exclusiveMaximum' in schema:
|
|
max_value = schema['exclusiveMaximum'] - 1
|
|
|
|
out = ["("]
|
|
_generate_min_max_int(min_value, max_value, out)
|
|
out.append(") space")
|
|
return self._add_rule(rule_name, ''.join(out))
|
|
|
|
elif (schema_type == 'object') or (len(schema) == 0):
|
|
return self._add_rule(rule_name, self._add_primitive('object', PRIMITIVE_RULES['object']))
|
|
|
|
elif schema_type is None and isinstance(schema, dict):
|
|
# No type constraint and no recognized structural keywords (e.g. {"description": "..."}).
|
|
# Per JSON Schema semantics this is equivalent to {} and accepts any value.
|
|
return self._add_rule(rule_name, self._add_primitive('value', PRIMITIVE_RULES['value']))
|
|
|
|
else:
|
|
assert schema_type in PRIMITIVE_RULES, f'Unrecognized schema: {schema}'
|
|
# TODO: support minimum, maximum, exclusiveMinimum, exclusiveMaximum at least for zero
|
|
return self._add_primitive('root' if rule_name == 'root' else schema_type, PRIMITIVE_RULES[schema_type])
|
|
|
|
def _add_primitive(self, name: str, rule: BuiltinRule):
|
|
n = self._add_rule(name, rule.content)
|
|
|
|
for dep in rule.deps:
|
|
dep_rule = PRIMITIVE_RULES.get(dep) or STRING_FORMAT_RULES.get(dep)
|
|
assert dep_rule, f'Rule {dep} not known'
|
|
if dep not in self._rules:
|
|
self._add_primitive(dep, dep_rule)
|
|
return n
|
|
|
|
def _build_object_rule(self, properties: List[Tuple[str, Any]], required: Set[str], name: str, additional_properties: Optional[Union[bool, Any]]):
|
|
prop_order = self._prop_order
|
|
# sort by position in prop_order (if specified) then by original order
|
|
sorted_props = [kv[0] for _, kv in sorted(enumerate(properties), key=lambda ikv: (prop_order.get(ikv[1][0], len(prop_order)), ikv[0]))]
|
|
|
|
prop_kv_rule_names = {}
|
|
for prop_name, prop_schema in properties:
|
|
prop_rule_name = self.visit(prop_schema, f'{name}{"-" if name else ""}{prop_name}')
|
|
prop_kv_rule_names[prop_name] = self._add_rule(
|
|
f'{name}{"-" if name else ""}{prop_name}-kv',
|
|
fr'{self._format_literal(json.dumps(prop_name))} space ":" space {prop_rule_name}'
|
|
)
|
|
required_props = [k for k in sorted_props if k in required]
|
|
optional_props = [k for k in sorted_props if k not in required]
|
|
|
|
if additional_properties is not None and additional_properties != False:
|
|
sub_name = f'{name}{"-" if name else ""}additional'
|
|
value_rule = self.visit(additional_properties, f'{sub_name}-value') if isinstance(additional_properties, dict) else \
|
|
self._add_primitive('value', PRIMITIVE_RULES['value'])
|
|
key_rule = self._add_primitive('string', PRIMITIVE_RULES['string']) if not sorted_props \
|
|
else self._add_rule(f'{sub_name}-k', self._not_strings(sorted_props))
|
|
|
|
prop_kv_rule_names["*"] = self._add_rule(
|
|
f'{sub_name}-kv',
|
|
f'{key_rule} ":" space {value_rule}'
|
|
)
|
|
optional_props.append("*")
|
|
|
|
rule = '"{" space '
|
|
rule += ' "," space '.join(prop_kv_rule_names[k] for k in required_props)
|
|
|
|
if optional_props:
|
|
rule += ' ('
|
|
if required_props:
|
|
rule += ' "," space ( '
|
|
|
|
def get_recursive_refs(ks, first_is_optional):
|
|
[k, *rest] = ks
|
|
kv_rule_name = prop_kv_rule_names[k]
|
|
comma_ref = f'( "," space {kv_rule_name} )'
|
|
if first_is_optional:
|
|
res = comma_ref + ('*' if k == '*' else '?')
|
|
else:
|
|
res = kv_rule_name + (' ' + comma_ref + "*" if k == '*' else '')
|
|
if len(rest) > 0:
|
|
res += ' ' + self._add_rule(
|
|
f'{name}{"-" if name else ""}{k}-rest',
|
|
get_recursive_refs(rest, first_is_optional=True)
|
|
)
|
|
return res
|
|
|
|
rule += ' | '.join(
|
|
get_recursive_refs(optional_props[i:], first_is_optional=False)
|
|
for i in range(len(optional_props))
|
|
)
|
|
if required_props:
|
|
rule += ' )'
|
|
rule += ' )?'
|
|
|
|
rule += ' "}" space'
|
|
|
|
return rule
|
|
|
|
def format_grammar(self):
|
|
return '\n'.join(
|
|
f'{name} ::= {rule}'
|
|
for name, rule in sorted(self._rules.items(), key=lambda kv: kv[0])
|
|
)
|
|
|
|
|
|
def main(args_in = None):
|
|
parser = argparse.ArgumentParser(
|
|
description='''
|
|
Generates a grammar (suitable for use in ./llama-cli) that produces JSON conforming to a
|
|
given JSON schema. Only a subset of JSON schema features are supported; more may be
|
|
added in the future.
|
|
''',
|
|
)
|
|
parser.add_argument(
|
|
'--prop-order',
|
|
default=[],
|
|
type=lambda s: s.split(','),
|
|
help='''
|
|
comma-separated property names defining the order of precedence for object properties;
|
|
properties not specified here are given lower precedence than those that are, and
|
|
are kept in their original order from the schema. Required properties are always
|
|
given precedence over optional properties.
|
|
'''
|
|
)
|
|
parser.add_argument(
|
|
'--allow-fetch',
|
|
action='store_true',
|
|
default=False,
|
|
help='Whether to allow fetching referenced schemas over HTTPS')
|
|
parser.add_argument(
|
|
'--dotall',
|
|
action='store_true',
|
|
default=False,
|
|
help='Whether to treat dot (".") as matching all chars including line breaks in regular expression patterns')
|
|
parser.add_argument(
|
|
'--raw-pattern',
|
|
action='store_true',
|
|
default=False,
|
|
help='Treats string patterns as raw patterns w/o quotes (or quote escapes)')
|
|
|
|
parser.add_argument('schema', help='file containing JSON schema ("-" for stdin)')
|
|
args = parser.parse_args(args_in)
|
|
|
|
if args.schema.startswith('https://'):
|
|
url = args.schema
|
|
import requests
|
|
schema = requests.get(url).json()
|
|
elif args.schema == '-':
|
|
url = 'stdin'
|
|
schema = json.load(sys.stdin)
|
|
else:
|
|
url = f'file://{args.schema}'
|
|
with open(args.schema) as f:
|
|
schema = json.load(f)
|
|
converter = SchemaConverter(
|
|
prop_order={name: idx for idx, name in enumerate(args.prop_order)},
|
|
allow_fetch=args.allow_fetch,
|
|
dotall=args.dotall,
|
|
raw_pattern=args.raw_pattern)
|
|
schema = converter.resolve_refs(schema, url)
|
|
converter.visit(schema, '')
|
|
print(converter.format_grammar())
|
|
|
|
|
|
if __name__ == '__main__':
|
|
main()
|