Skip to main content

​
StructuredOutputHandler

class StructuredOutputHandler:
Handler for generating prompts and extracting structured output from agent responses. This handler provides functionality to:
  • Generate prompts that guide agents to produce structured output
  • Extract structured data from agent responses using regex patterns
  • Provide fallback mechanisms when extraction fails
  • Support the existing structured output schemas used in workforce.py

​
generate_structured_prompt

def generate_structured_prompt(
    base_prompt: str,
    schema: Type[BaseModel],
    examples: Optional[List[Dict[str, Any]]] = None,
    additional_instructions: Optional[str] = None
):
Generate a prompt that guides agents to produce structured output. Parameters:
  • base_prompt (str): The base prompt content.
  • schema (Type[BaseModel]): The Pydantic model schema for the expected output.
  • examples (Optional[List[Dict[str, Any]]]): Optional examples of valid output.
  • additional_instructions (Optional[str]): Additional instructions for output formatting.
Returns: str: The enhanced prompt with structured output instructions.

​
extract_json

def extract_json(text: str, schema: Optional[Type[BaseModel]] = None):
Extract JSON data from text using multiple patterns. Parameters:
  • text (str): The text containing JSON data.
  • schema (Optional[Type[BaseModel]]): Optional schema for targeted extraction.
Returns: Optional[Dict[str, Any]]: Extracted JSON data or None if extraction fails.

​
_extract_with_schema_patterns

def _extract_with_schema_patterns(text: str, schema: Type[BaseModel]):
Extract data using schema-specific patterns. Parameters:
  • text (str): The text to extract from.
  • schema (Type[BaseModel]): The schema to use for extraction.
Returns: Optional[Dict[str, Any]]: Extracted data or None.

​
parse_structured_response

def parse_structured_response(
    response_text: str,
    schema: Type[BaseModel],
    fallback_values: Optional[Dict[str, Any]] = None
):
Parse agent response into structured data with fallback support. Parameters:
  • response_text (str): The agent’s response text.
  • schema (Type[BaseModel]): The expected schema.
  • fallback_values (Optional[Dict[str, Any]]): Fallback values to use if parsing fails.
Returns: Union[BaseModel, Dict[str, Any]]: Parsed data as schema instance or fallback dictionary.

​
_fix_common_issues

def _fix_common_issues(data: Dict[str, Any], schema: Type[BaseModel]):
Attempt to fix common validation issues in extracted data. Parameters:
  • data (Dict[str, Any]): The extracted data.
  • schema (Type[BaseModel]): The target schema.
Returns: Optional[Dict[str, Any]]: Fixed data or None if unfixable.

​
_create_default_instance

def _create_default_instance(schema: Type[BaseModel]):
Create a default instance of the schema with minimal required fields. Parameters:
  • schema (Type[BaseModel]): The schema to instantiate.
Returns: BaseModel: Default instance of the schema.

​
validate_response

def validate_response(
    response: Union[BaseModel, Dict[str, Any]],
    schema: Type[BaseModel]
):
Validate that a response conforms to the expected schema. Parameters:
  • response: The response to validate.
  • schema (Type[BaseModel]): The expected schema.
Returns: bool: True if valid, False otherwise.

​
create_fallback_response

def create_fallback_response(
    schema: Type[BaseModel],
    error_message: str,
    context: Optional[Dict[str, Any]] = None
):
Create a fallback response for a given schema with error context. Parameters:
  • schema (Type[BaseModel]): The schema to create a response for.
  • error_message (str): The error message to include.
  • context (Optional[Dict[str, Any]]): Additional context for the fallback.
Returns: BaseModel: A valid instance of the schema with fallback values.