feat: LLM-native structured output via JSON schema enforcement (#1037)

Thread existing JSON schemas from prompt definitions through the
text-completion service to LLM backends' native structured output
APIs. When a prompt has response-type "json" and a strict-mode
compatible schema, the LLM constrains token selection at the logit
level to guarantee schema-valid output.

Wire-level changes:
- Add response_format and schema fields to TextCompletionRequest
- Update translator to encode/decode new fields
- Pass new fields through LlmService, TextCompletionClient, and
  PromptManager

Runtime schema compatibility checker:
- New is_strict_mode_compatible() utility validates schemas against
  LLM provider constraints (additionalProperties, required fields,
  no unsupported constraints, no open-ended objects)
- Per-prompt eligibility decision: compliant schemas use structured
  output, non-compliant schemas fall back to free-text + post-hoc
  validation

LLM backend implementations:
- OpenAI: response_format with json_schema, variant-aware top-level
  array rejection (openai variant blocks, llama/vllm variants allow)
- New vllm variant for the OpenAI backend
- vLLM (dedicated): response_format in raw HTTP body
- Ollama: format=<schema> parameter
- Claude: tool-use trick (forced tool call with schema as input_schema)
- Mistral: native json_schema response_format
- Llamafile, LM Studio: OpenAI SDK response_format
- Azure OpenAI: AzureOpenAI SDK response_format
- Azure serverless: response_format in raw HTTP body
- TGI: response_format in raw HTTP body
- VertexAI Gemini: response_mime_type + response_schema
- VertexAI Claude: tool-use trick
- Google AI Studio: response_mime_type + response_schema
- Bedrock, Cohere: signature-only (no structured output yet)

Post-hoc jsonschema.validate() retained as defence-in-depth.

Tech spec added: docs/tech-specs/structured-output.md

Update tests
This commit is contained in:
cybermaggedon 2026-07-10 15:28:56 +01:00 committed by GitHub
parent f106ae2103
commit 9136526863
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
27 changed files with 1089 additions and 71 deletions

View file

@ -49,4 +49,5 @@ from . keyword_index_client import KeywordIndexClientSpec, KeywordIndexClient
from . row_embeddings_query_client import RowEmbeddingsQueryClientSpec
from . collection_config_handler import CollectionConfigHandler
from . audit_publisher import AuditPublisher
from . schema_compatibility import is_strict_mode_compatible

View file

@ -126,6 +126,8 @@ class LlmService(FlowProcessor):
# Check if streaming is requested and supported
streaming = getattr(request, 'streaming', False)
response_format = getattr(request, 'response_format', None)
schema = getattr(request, 'schema', None)
if streaming and self.supports_streaming():
@ -136,7 +138,8 @@ class LlmService(FlowProcessor):
).time():
async for chunk in self.generate_content_stream(
request.system, request.prompt, model, temperature
request.system, request.prompt, model, temperature,
response_format=response_format, schema=schema,
):
await flow("response").send(
TextCompletionResponse(
@ -159,7 +162,8 @@ class LlmService(FlowProcessor):
).time():
response = await self.generate_content(
request.system, request.prompt, model, temperature
request.system, request.prompt, model, temperature,
response_format=response_format, schema=schema,
)
await flow("response").send(
@ -215,7 +219,10 @@ class LlmService(FlowProcessor):
"""
return False
async def generate_content_stream(self, system, prompt, model=None, temperature=None):
async def generate_content_stream(
self, system, prompt, model=None, temperature=None,
response_format=None, schema=None,
):
"""
Override in subclass to implement streaming.
Should yield LlmChunk objects.

View file

@ -0,0 +1,90 @@
import logging
logger = logging.getLogger(__name__)
def is_strict_mode_compatible(schema):
"""
Check whether a JSON schema is compatible with LLM structured-output
strict mode. Returns True if the schema can be passed directly to
providers like OpenAI, vLLM, etc.
"""
if schema is None:
return False
try:
_check_node(schema)
return True
except _IncompatibleSchema as e:
logger.debug("Schema not strict-mode compatible: %s", e)
return False
class _IncompatibleSchema(Exception):
pass
def _check_node(node):
if not isinstance(node, dict):
return
node_type = node.get("type")
if node_type == "object" or (
node_type is None and "properties" in node
):
_check_object(node)
if node_type == "array":
items = node.get("items")
if items:
_check_node(items)
for keyword in ("oneOf", "anyOf", "allOf"):
for child in node.get(keyword, []):
_check_node(child)
_check_unsupported_constraints(node)
def _check_object(node):
props = node.get("properties")
if props is None:
raise _IncompatibleSchema(
"object without properties (open-ended)"
)
if node.get("additionalProperties") is not False:
raise _IncompatibleSchema(
"object missing additionalProperties: false"
)
required = set(node.get("required", []))
for key in props:
if key not in required:
raise _IncompatibleSchema(
f"property '{key}' not in required"
)
for value in props.values():
_check_node(value)
UNSUPPORTED_KEYWORDS = {
"minimum", "maximum", "exclusiveMinimum", "exclusiveMaximum",
"minLength", "maxLength", "pattern",
"minItems", "maxItems",
"minProperties", "maxProperties",
}
def _check_unsupported_constraints(node):
found = UNSUPPORTED_KEYWORDS & node.keys()
if found:
raise _IncompatibleSchema(
f"unsupported constraints: {', '.join(sorted(found))}"
)

View file

@ -14,11 +14,15 @@ class TextCompletionResult:
class TextCompletionClient(RequestResponse):
async def text_completion(self, system, prompt, timeout=600):
async def text_completion(
self, system, prompt, timeout=600,
response_format=None, schema=None,
):
resp = await self.request(
TextCompletionRequest(
system = system, prompt = prompt, streaming = False
system=system, prompt=prompt, streaming=False,
response_format=response_format, schema=schema,
),
timeout=timeout
)
@ -35,6 +39,7 @@ class TextCompletionClient(RequestResponse):
async def text_completion_stream(
self, system, prompt, handler, timeout=600,
response_format=None, schema=None,
):
"""
Streaming text completion. `handler` is an async callable invoked
@ -54,7 +59,8 @@ class TextCompletionClient(RequestResponse):
final = await self.request(
TextCompletionRequest(
system = system, prompt = prompt, streaming = True
system=system, prompt=prompt, streaming=True,
response_format=response_format, schema=schema,
),
recipient=on_chunk,
timeout=timeout,

View file

@ -10,14 +10,21 @@ class TextCompletionRequestTranslator(MessageTranslator):
return TextCompletionRequest(
system=data["system"],
prompt=data["prompt"],
streaming=data.get("streaming", False)
streaming=data.get("streaming", False),
response_format=data.get("response_format"),
schema=data.get("schema"),
)
def encode(self, obj: TextCompletionRequest) -> Dict[str, Any]:
return {
result = {
"system": obj.system,
"prompt": obj.prompt
"prompt": obj.prompt,
}
if obj.response_format is not None:
result["response_format"] = obj.response_format
if obj.schema is not None:
result["schema"] = obj.schema
return result
class TextCompletionResponseTranslator(MessageTranslator):

View file

@ -11,7 +11,9 @@ from ..core.primitives import Error
class TextCompletionRequest:
system: str = ""
prompt: str = ""
streaming: bool = False # Default false for backward compatibility
streaming: bool = False
response_format: str | None = None
schema: dict | None = None
@dataclass
class TextCompletionResponse: