07. Multi-Tool Orchestration

Multi-tool orchestration manages sequences where one tool call depends on another's output. The model decides which tools to call and in what order based on the user goal.

Implement a loop that continues until the model produces a text response:

def multi_tool_orchestration(
    model: str,
    messages: list[dict],
    tools: list[dict],
    registry: ToolRegistry,
    max_iterations: int = 10
) -> str:
    iteration = 0
    
    while iteration < max_iterations:
        iteration += 1
        
        # Get model response
        response = call_ollama_with_tools(model, messages, tools)
        
        # Extract tool calls
        tool_calls = extract_tool_calls(response)
        
        if not tool_calls:
            # Model returned natural language response
            return response["message"]["content"]
        
        # Execute each tool call
        for call in tool_calls:
            schema = registry.get_schema(call["name"])
            
            # Record the call
            messages.append({
                "role": "assistant",
                "content": None,
                "tool_calls": [{
                    "id": f"call_{call['name']}_{iteration}",
                    "type": "function",
                    "function": {
                        "name": call["name"],
                        "arguments": json.dumps(call["arguments"])
                    }
                }]
            })
            
            # Execute and record result
            result = execute_with_retry(call["name"], call["arguments"], schema)
            
            messages.append({
                "role": "tool",
                "tool_call_id": f"call_{call['name']}_{iteration}",
                "content": json.dumps(result)
            })
    
    return "Maximum iterations reached without completion"

The max_iterations guard prevents infinite loops when models get stuck calling tools without making progress. Set this based on typical task complexity—simple queries need 2-3 iterations, complex reasoning may need 8-10.

State management matters for multi-tool sequences. Track tool call history and intermediate results:

class OrchestrationState:
    def __init__(self):
        self.messages = []
        self.execution_history = []
        self.final_result = None
    
    def add_tool_result(self, name: str, arguments: dict, result: dict):
        self.execution_history.append({
            "tool": name,
            "arguments": arguments,
            "result": result,
            "success": result.get("success", False)
        })
    
    def get_successful_tools(self) -> list[str]:
        return [h["tool"] for h in self.execution_history if h["success"]]

Dependency tracking handles tools that require previous tool outputs:

def resolve_argument_dependencies(
    arguments: dict,
    execution_history: list[dict]
) -> dict:
    resolved = {}
    
    for key, value in arguments.items():
        if isinstance(value, str) and value.startswith("$"):
            # Reference to previous tool output
            tool_index = int(value.split("[")[1].split("]")[0])
            field = value.split(".")[1] if "." in value else "result"
            
            if tool_index < len(execution_history):
                resolved[key] = execution_history[tool_index]["result"].get(field)
            else:
                raise ValueError(f"Invalid tool reference: {value}")
        else:
            resolved[key] = value
    
    return resolved