Function execute_pattern_chain

scripts/python_ui/streamlit.py:1141–1225

Execute a sequence of patterns in a chain, passing output from each to the next. Args: patterns_sequence: List of pattern names to execute in sequence initial_input: Initial input text to start the chain Returns: Dict containing results from each stage of the chain

(patterns_sequence: List[str], initial_input: str)

def execute_pattern_chain(patterns_sequence: List[str], initial_input: str) -> Dict:
    """Execute a sequence of patterns in a chain, passing output from each to the next.

    Args:
        patterns_sequence: List of pattern names to execute in sequence
        initial_input: Initial input text to start the chain

    Returns:
        Dict containing results from each stage of the chain
    """
    logger.info(
        f"Starting pattern chain execution with {len(patterns_sequence)} patterns"
    )
    chain_results = {
        "sequence": patterns_sequence,
        "stages": [],
        "final_output": None,
        "metadata": {
            "timestamp": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
            "success": False,
        },
    }

    current_input = initial_input

    try:
        for i, pattern in enumerate(patterns_sequence, 1):
            logger.info(f"Chain Stage {i}: Executing pattern '{pattern}'")
            stage_result = {
                "pattern": pattern,
                "input": current_input,
                "output": None,
                "success": False,
                "error": None,
            }

            try:
                cmd = ["fabric", "--pattern", pattern]
                result = run(
                    cmd, input=current_input, capture_output=True, text=True, check=True
                )
                output = result.stdout.strip()

                if output:
                    stage_result["output"] = output
                    stage_result["success"] = True
                    current_input = output  # Use this output as input for next pattern
                    logger.debug(f"Stage {i} completed successfully")
                else:
                    stage_result["error"] = "Pattern generated no output"
                    logger.warning(f"Pattern {pattern} generated no output")

            except CalledProcessError as e:
                error_msg = f"Error executing pattern: {e.stderr.strip()}"
                stage_result["error"] = error_msg
                logger.error(error_msg)
                break