rob/CmdForge
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Implement structured output for prompt steps (Phase 1)

Browse Source 
Add schema validation and retry logic for AI responses:

- Add output_schema, max_retries, plain_text fields to PromptStep
- Add DEFAULT_OUTPUT_SCHEMA with output/reasoning fields
- Implement append_schema_instructions() for prompt augmentation
- Implement validate_schema() using jsonschema library
- Add retry logic with error feedback on validation failure
- Enhance substitute_variables() for nested field access ({var.field})
- Add jsonschema to dependencies
- Update GUI dialog with plain_text checkbox and max_retries spinner
- Add comprehensive tests for structured output and nested access

Tools using plain_text=True bypass validation (current behavior).
Tools without plain_text get structured JSON output with schema enforcement.

Migration Phase 1 complete. Next: Apply plain_text=True to all 72 tools.

Co-Authored-By: Claude Opus 4.5 <[email protected]>
This commit is contained in:
rob
2026-02-17 17:34:48 -04:00
co-authored by Claude Opus 4.5
parent f980fe05f1
commit 8581775002
6 changed files with 566 additions and 46 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/cmdforge/runner.py
+254 -32
View File
@@ -1,6 +1,7 @@
"""Tool execution engine."""
import argparse
import json
import sys
from pathlib import Path
from typing import Optional
@@ -16,6 +17,80 @@ from .profiles import load_profile
# Maximum recursion depth for nested tool calls
MAX_TOOL_DEPTH = 10
# Default schema for structured output
# Separates AI reasoning from actual output
DEFAULT_OUTPUT_SCHEMA = {
"type": "object",
"properties": {
"output": {
"type": "string",
"description": "The actual response content"
},
"reasoning": {
"type": "string",
"description": "Internal thoughts, analysis, or step-by-step reasoning"
}
},
"required": ["output"]
}
class SchemaValidationError(Exception):
"""Raised when response doesn't match expected schema."""
pass
def validate_schema(data: dict, schema: dict) -> None:
"""
Validate data against a JSON schema.
Args:
data: Parsed JSON data
schema: JSON schema to validate against
Raises:
SchemaValidationError: If validation fails
"""
try:
from jsonschema import validate, ValidationError
validate(instance=data, schema=schema)
except ValidationError as e:
# Extract path info for clearer error
path = ".".join(str(p) for p in e.absolute_path) if e.absolute_path else "root"
raise SchemaValidationError(f"Field '{path}': {e.message}")
except ImportError:
# jsonschema not installed - skip validation with warning
print("Warning: jsonschema not installed, skipping schema validation", file=sys.stderr)
def append_schema_instructions(prompt: str, schema: dict) -> str:
"""
Append schema instructions to a prompt.
Args:
prompt: Original prompt text
schema: JSON schema the response must match
Returns:
Augmented prompt with schema instructions
"""
schema_json = json.dumps(schema, indent=2)
return f"""{prompt}
---
RESPONSE FORMAT:
You must respond with ONLY valid JSON matching this exact schema:
{schema_json}
CRITICAL RULES:
- Output ONLY the JSON object, no other text before or after
- Put any thinking, reasoning, or analysis in the "reasoning" field
- Put your actual answer/response in the "output" field
- Do not include markdown code fences
- Do not include any preamble like "Here is my response:"
"""
def check_system_dependencies(tool: Tool) -> list:
"""
@@ -114,14 +189,52 @@ def check_dependencies(tool: Tool, checked: set = None) -> list[str]:
return missing
def _get_nested_value(obj, path_parts: list):
"""
Navigate nested dict/JSON to get a value.
Args:
obj: Starting object (dict or JSON string)
path_parts: List of keys to navigate
Returns:
Value at path, or None if not found
"""
value = obj
for part in path_parts:
if isinstance(value, dict):
value = value.get(part)
elif isinstance(value, str):
# Try parsing as JSON
try:
parsed = json.loads(value)
if isinstance(parsed, dict):
value = parsed.get(part)
else:
return None
except json.JSONDecodeError:
return None
else:
return None
if value is None:
return None
return value
def substitute_variables(template: str, variables: dict, warn_non_scalar: bool = False) -> str:
"""
Substitute {variable} placeholders in a template.
Supports escaping: use {{ for literal { and }} for literal }
Supports:
- Simple: {varname}
- Nested: {varname.field} or {varname.field.subfield}
- Settings: {settings.key}
- Escaping: {{ for literal { and }} for literal }
Also supports {settings.key} syntax for accessing top-level scalar
values from the settings dict.
For nested access, if the variable is a JSON string, it will be parsed.
Args:
template: String with {var} placeholders
@@ -137,6 +250,8 @@ def substitute_variables(template: str, variables: dict, warn_non_scalar: bool =
'Use {braces}'
>>> substitute_variables("Backend: {settings.backend}", {"settings": {"backend": "piper"}})
'Backend: piper'
>>> substitute_variables("Answer: {result.output}", {"result": '{"output": "Hello"}'})
'Answer: Hello'
"""
import re
@@ -147,34 +262,60 @@ def substitute_variables(template: str, variables: dict, warn_non_scalar: bool =
# First, replace escaped braces with placeholders
result = template.replace("{{", ESCAPE_OPEN).replace("}}", ESCAPE_CLOSE)
# Handle settings.key syntax (top-level scalar values only)
settings = variables.get("settings", {})
if settings and isinstance(settings, dict):
def replace_settings(match):
key = match.group(1)
# Pattern for {var} or {var.field.subfield}
pattern = r'\{([a-zA-Z_][a-zA-Z0-9_]*(?:\.[a-zA-Z_][a-zA-Z0-9_]*)*)\}'
def replace_var(match):
path = match.group(1)
parts = path.split('.')
# Get base variable
base_name = parts[0]
# Check if variable exists in dict (distinguish None value from missing key)
if base_name not in variables:
return match.group(0) # Leave unchanged for missing variables
base_value = variables[base_name]
# If no nested parts, simple substitution
if len(parts) == 1:
if base_name == "settings":
# Settings dict - leave unchanged (handled specially)
return match.group(0)
# Replace with empty string for None, str() for other values
return "" if base_value is None else str(base_value)
# Handle nested access
if base_name == "settings":
# Settings dict - only allow scalar values in templates
settings = base_value if isinstance(base_value, dict) else {}
key = parts[1] # Only support one level: settings.key
if key in settings:
value = settings[key]
if isinstance(value, (str, int, float, bool)):
return str(value) if value is not None else ""
else:
# Non-scalar value - leave placeholder, warn if requested
if warn_non_scalar:
print(
f"Warning: {{settings.{key}}} is not a scalar value, "
f"access it in code steps instead",
file=sys.stderr
)
return match.group(0) # Return unchanged
return match.group(0) # Key not found, leave unchanged
return match.group(0)
return match.group(0)
result = re.sub(r'\{settings\.([^}]+)\}', replace_settings, result)
# Navigate nested path for other variables
value = _get_nested_value(base_value, parts[1:])
if value is None:
return match.group(0) # Leave unchanged if not found
# Now do regular variable substitution (skip 'settings' since we handled it)
for name, value in variables.items():
if name == "settings":
continue # Already handled above
# Use 'is not None' to preserve falsey values like 0 and False
result = result.replace(f"{{{name}}}", "" if value is None else str(value))
# Convert to string for template
if isinstance(value, (dict, list)):
return json.dumps(value)
return str(value)
result = re.sub(pattern, replace_var, result)
# Finally, restore escaped braces as single braces
result = result.replace(ESCAPE_OPEN, "{").replace(ESCAPE_CLOSE, "}")
@@ -198,7 +339,12 @@ def execute_prompt_step(
Returns:
Tuple of (output_value, success)
If plain_text=True, returns raw provider output.
Otherwise, enforces structured JSON output with schema validation.
"""
import re
# Build prompt with variable substitution
prompt = substitute_variables(step.prompt, variables, warn_non_scalar=verbose)
@@ -209,25 +355,101 @@ def execute_prompt_step(
# Prepend system prompt to user prompt
prompt = f"{profile.system_prompt}\n\n---\n\n{prompt}"
# Call provider (support variable substitution in provider name, e.g. {settings.provider})
# Determine provider
provider = provider_override or substitute_variables(step.provider, variables, warn_non_scalar=verbose)
if provider.lower() == "mock":
result = mock_provider(prompt)
else:
result = call_provider(provider, prompt)
# Plain text mode - bypass structured output
if step.plain_text:
if provider.lower() == "mock":
result = mock_provider(prompt)
else:
result = call_provider(provider, prompt)
if not result.success:
print(f"Error in prompt step: {result.error}", file=sys.stderr)
return "", False
if not result.success:
print(f"Error in prompt step: {result.error}", file=sys.stderr)
return "", False
text = result.text
if step.strip_fences:
import re
text = re.sub(r'^```\w*\n', '', text.strip())
text = re.sub(r'\n```\s*$', '', text)
text = result.text
if step.strip_fences:
text = re.sub(r'^```\w*\n', '', text.strip())
text = re.sub(r'\n```\s*$', '', text)
return text, True
return text, True
# Structured output mode - enforce JSON with schema validation
schema = step.output_schema or DEFAULT_OUTPUT_SCHEMA
# Augment prompt with schema instructions
augmented_prompt = append_schema_instructions(prompt, schema)
max_attempts = step.max_retries + 1
last_response = None
last_error = None
for attempt in range(max_attempts):
current_prompt = augmented_prompt
if attempt > 0 and last_response:
# Retry with error feedback
current_prompt = f"""{augmented_prompt}
Your previous response was:
{last_response[:1000]}{"..." if len(last_response) > 1000 else ""}
This failed validation: {last_error}
Please try again with valid JSON matching the schema exactly."""
if verbose:
if attempt > 0:
print(f"[verbose] Retry {attempt}/{step.max_retries} after validation failure", file=sys.stderr)
# Call provider
if provider.lower() == "mock":
result = mock_provider(current_prompt)
else:
result = call_provider(provider, current_prompt)
if not result.success:
print(f"Error in prompt step: {result.error}", file=sys.stderr)
return "", False
last_response = result.text
# Strip markdown code fences if present (common model behavior)
text = result.text.strip()
if text.startswith("```"):
text = re.sub(r'^```\w*\n', '', text)
text = re.sub(r'\n```\s*$', '', text)
# Try to parse as JSON
try:
parsed = json.loads(text)
except json.JSONDecodeError as e:
last_error = f"Invalid JSON: {e}"
if attempt == max_attempts - 1:
print(f"Prompt step failed after {step.max_retries} retry(s): {last_error}", file=sys.stderr)
if verbose:
print(f"[verbose] Last response: {last_response[:500]}...", file=sys.stderr)
return "", False
continue
# Validate against schema
try:
validate_schema(parsed, schema)
except SchemaValidationError as e:
last_error = str(e)
if attempt == max_attempts - 1:
print(f"Prompt step failed after {step.max_retries} retry(s): {last_error}", file=sys.stderr)
if verbose:
print(f"[verbose] Parsed response: {json.dumps(parsed)[:500]}...", file=sys.stderr)
return "", False
continue
# Success - return normalized JSON string
return json.dumps(parsed), True
# Should not reach here, but just in case
return "", False
def execute_code_step(