[Fix & Style] Refactor the grammar backend to reduce human errors and improve readability (#4030)
This commit is contained in:
@@ -57,7 +57,7 @@ class XGrammarGrammar(BaseGrammarObject):
|
||||
def accept_token(self, token: int):
|
||||
assert self.matcher.accept_token(token)
|
||||
|
||||
def try_jump_forward(self, tokenizer) -> Tuple[List[int], str]:
|
||||
def try_jump_forward(self, tokenizer) -> Optional[Tuple[List[int], str]]:
|
||||
s = self.matcher.find_jump_forward_string()
|
||||
if s:
|
||||
return [], s
|
||||
@@ -128,55 +128,56 @@ class XGrammarGrammarBackend(BaseGrammarBackend):
|
||||
self.vocab_size = vocab_size
|
||||
self.override_stop_tokens = override_stop_tokens
|
||||
|
||||
def init_value_impl(self, key: Tuple[str, str]) -> XGrammarGrammar:
|
||||
|
||||
key_type, key_string = key
|
||||
if key_type == "json":
|
||||
try:
|
||||
if key_string == "$$ANY$$":
|
||||
ctx = self.grammar_compiler.compile_builtin_json_grammar()
|
||||
else:
|
||||
ctx = self.grammar_compiler.compile_json_schema(schema=key_string)
|
||||
except RuntimeError as e:
|
||||
logging.warning(
|
||||
f"Skip invalid json_schema: json_schema={key_string}, {e=}"
|
||||
)
|
||||
return None
|
||||
elif key_type == "ebnf":
|
||||
try:
|
||||
ctx = self.grammar_compiler.compile_grammar(key_string)
|
||||
except RuntimeError as e:
|
||||
logging.warning(f"Skip invalid ebnf: ebnf={key_string}, {e=}")
|
||||
return None
|
||||
elif key_type == "regex":
|
||||
try:
|
||||
ctx = self.grammar_compiler.compile_regex(key_string)
|
||||
except RuntimeError as e:
|
||||
logging.warning(f"Skip invalid regex: regex={key_string}, {e=}")
|
||||
return None
|
||||
elif key_type == "structural_tag":
|
||||
try:
|
||||
structural_tag = json.loads(key_string)
|
||||
tags = [
|
||||
StructuralTagItem(
|
||||
begin=structure["begin"],
|
||||
schema=json.dumps(structure["schema"]),
|
||||
end=structure["end"],
|
||||
)
|
||||
for structure in structural_tag["structures"]
|
||||
]
|
||||
ctx = self.grammar_compiler.compile_structural_tag(
|
||||
tags, structural_tag["triggers"]
|
||||
)
|
||||
except RuntimeError as e:
|
||||
logging.warning(f"Skip invalid regex: regex={key_string}, {e=}")
|
||||
return None
|
||||
else:
|
||||
raise ValueError(f"Invalid key_type: {key_type}")
|
||||
|
||||
def _from_context(self, ctx: CompiledGrammar) -> XGrammarGrammar:
|
||||
matcher = GrammarMatcher(ctx, max_rollback_tokens=MAX_ROLLBACK_TOKENS)
|
||||
return XGrammarGrammar(matcher, self.vocab_size, ctx, self.override_stop_tokens)
|
||||
|
||||
def dispatch_json(self, key_string: str) -> Optional[XGrammarGrammar]:
|
||||
try:
|
||||
if key_string == "$$ANY$$":
|
||||
ctx = self.grammar_compiler.compile_builtin_json_grammar()
|
||||
else:
|
||||
ctx = self.grammar_compiler.compile_json_schema(schema=key_string)
|
||||
except RuntimeError as e:
|
||||
logging.warning(f"Skip invalid json_schema: json_schema={key_string}, {e=}")
|
||||
return None
|
||||
return self._from_context(ctx)
|
||||
|
||||
def dispatch_ebnf(self, key_string: str) -> Optional[XGrammarGrammar]:
|
||||
try:
|
||||
ctx = self.grammar_compiler.compile_grammar(key_string)
|
||||
except RuntimeError as e:
|
||||
logging.warning(f"Skip invalid ebnf: ebnf={key_string}, {e=}")
|
||||
return None
|
||||
return self._from_context(ctx)
|
||||
|
||||
def dispatch_regex(self, key_string: str) -> Optional[XGrammarGrammar]:
|
||||
try:
|
||||
ctx = self.grammar_compiler.compile_regex(key_string)
|
||||
except RuntimeError as e:
|
||||
logging.warning(f"Skip invalid regex: regex={key_string}, {e=}")
|
||||
return None
|
||||
return self._from_context(ctx)
|
||||
|
||||
def dispatch_structural_tag(self, key_string: str) -> Optional[XGrammarGrammar]:
|
||||
try:
|
||||
structural_tag = json.loads(key_string)
|
||||
tags = [
|
||||
StructuralTagItem(
|
||||
begin=structure["begin"],
|
||||
schema=json.dumps(structure["schema"]),
|
||||
end=structure["end"],
|
||||
)
|
||||
for structure in structural_tag["structures"]
|
||||
]
|
||||
ctx = self.grammar_compiler.compile_structural_tag(
|
||||
tags, structural_tag["triggers"]
|
||||
)
|
||||
except RuntimeError as e:
|
||||
logging.warning(f"Skip invalid regex: regex={key_string}, {e=}")
|
||||
return None
|
||||
return self._from_context(ctx)
|
||||
|
||||
def reset(self):
|
||||
if self.grammar_compiler:
|
||||
self.grammar_compiler.clear_cache()
|
||||
|
||||
Reference in New Issue
Block a user