Function execute_step

src/main.py:831–940

Execute a single step in the execution plan. Handles step execution by: 1. Generating specialized system prompt for the step 2. Gathering context from dependent steps 3. Substituting variables in arguments 4. Executing the step with appropriate error handling Args:

(
    state: RuntimeState,
    step_id: StepID,
    llm: Optional[ChatOpenAI] = None,
    save_steps: bool = False,
    output_dir: str = './steps'
)

from langchain_core.output_parsers import StrOutputParser

async def execute_step(
    state: RuntimeState,
    step_id: StepID,
    llm: Optional[ChatOpenAI] = None,
    save_steps: bool = False,
    output_dir: str = './steps'
) -> Dict[str, Any]:
    """Execute a single step in the execution plan.
    
    Handles step execution by:
    1. Generating specialized system prompt for the step
    2. Gathering context from dependent steps
    3. Substituting variables in arguments
    4. Executing the step with appropriate error handling
    
    Args:
        state: Current runtime state
        step_id: ID of step to execute
        llm: Optional LLM instance to use
        save_steps: Whether to save step results to files
        output_dir: Directory to save step files
        
    Returns:
        Dict[str, Any]: Updates to state including step results
    """
    try:
        step_key = str(step_id)
        step_info = state["step_data"][step_key]
        logger.info(f"Executing step #{step_key}: {step_info.description}")
        
        # Create or use provided LLM instance
        step_llm = llm or create_llm() 
        
        # Generate specialized system prompt
        system_prompt = await generate_specialist_prompt(step_info.description)
        logger.debug(f"Generated system prompt for step {step_id}")
        
        # Build context from dependencies
        context = []
        agent_results = state.get("agent_results", {})
        
        for dep in step_info.depends_on:
            if str(dep) in agent_results:
                result = agent_results[str(dep)].get("result")
                context.append(f"Previous step {dep}: {result}")
            else:
                logger.warning(f"Missing result for dependency {dep}")
                context.append(f"Previous step {dep}: Not found")
        
        # Substitute variables in arguments
        step_args = substitute_variables(
            step_info.args,
            step_info.depends_on,
            agent_results
        )
        
        # Create and execute prompt
        prompt = ChatPromptTemplate.from_messages([