The AI Interview - Master AI/ML Interviews

Implement tool composition engine:

Composition Patterns:

Sequential: Output of A → Input of B
Parallel: Run A, B, C together, merge results
Conditional: If condition then A else B

Methods:

register_tool(name, func, input_schema, output_schema): Add tool
compose_sequential(name, chain, data_mapping): Chain tools
- data_mapping: how to route outputs to inputs
compose_parallel(name, tools, merge_strategy): Parallel execution
compose_conditional(name, condition, then, else_): Branching
execute_composition(name, input): Run composed workflow

Data Mapping: {'tool2_param': 'tool1_result.field'} extracts field from previous output

Merge Strategies:

'list': [result1, result2, ...]
'dict': {tool1: result1, ...}
'sum': sum(results)

Examples

Example 1:

Input:

engine = ToolCompositionEngine(); engine.register_tool('double', lambda x: x*2, {'x': 'int'}, {'result': 'int'}); 'double' in engine.tools

Output: True

Explanation: Tool registered successfully

Starter Code

class ToolCompositionEngine:
    """
    Engine for composing and chaining tools into complex workflows.
    """
    
    def __init__(self):
        self.tools = {}
        self.compositions = {}
    
    def register_tool(self, name, func, input_schema, output_schema):
        """Register tool with schemas for type checking"""
        # Your implementation here
        pass
    
    def compose_sequential(self, name, tool_chain, data_mapping):
        """
        Compose tools sequentially.
        tool_chain: ['tool1', 'tool2', ...]
        data_mapping: {'tool2_input': 'tool1_output.field'}
        """
        # Your implementation here
        pass
    
    def compose_parallel(self, name, tools, merge_strategy='list'):
        """
        Compose tools to run in parallel.
        merge_strategy: 'list', 'dict', 'sum', custom_fn
        """
        # Your implementation here
        pass
    
    def compose_conditional(self, name, condition_tool, then_branch, else_branch):
        """Compose if-then-else tool workflow"""
        # Your implementation here
        pass
    
    def execute_composition(self, name, initial_input):
        """Execute a composed tool chain"""
        # Your implementation here
        pass

Tool Composition and Chaining Engine

Examples

Starter Code