hamilton-dev-workflow

<!-- SPDX-License-Identifier: Apache-2.0 --> # Hamilton Development Workflow for Claude Code **TL;DR**: Build Hamilton DAGs systematically using DOT → Signatures → Validation → TDD implementation. ## Why This Workflow? This workflow optimizes for: - **Token efficiency**: DOT language is extremely concise - **Early validation**: Catch graph errors before implementation - **LLM-friendly**: Well-structured, incremental approach - **Testability**: Build complex systems node-by-node with tests - **Pragmatic typing**: Handle type annotations strategically ## The 5-Step Workflow ### Step 1: Natural Language → DOT Graph **Convert user requirements into DOT language representation** DOT language is: - Simple and well-understood by LLMs - Extremely token-efficient - Perfect for representing DAG structure - Easy to validate visually **Example:** ``` User: "Load CSV data, clean nulls, calculate 7-day rolling average, output top 10" DOT representation: digraph hamilton_dag { // Inputs data_path [shape=box, color=blue] // Pipeline (arrows show data flow: upstream -> downstream) data_path -> raw_data raw_data -> cleaned_data cleaned_data -> rolling_7d rolling_7d -> top_10 } ``` **Best Practices:** - Use descriptive node names (snake_case) - Mark inputs with `[shape=box, color=blue]` - Add comments for clarity - Keep it simple - just the graph structure ### Step 2: DOT Graph → Function Signatures + Docstrings **Convert DOT nodes into Python function signatures** This step is mechanical and can be automated. For each node: 1. Create a function with the node name 2. Add parameters based on incoming edges 3. Add return type annotation (use `Any` if uncertain) 4. Write a clear docstring explaining what the node computes **Example:** ```python """Data processing pipeline.""" from typing import Any import pandas as pd def raw_data(data_path: str) -> pd.DataFrame: """Load raw data from CSV file. :param data_path: Path to CSV file :return: Raw DataFrame """ pass def cleaned_data(raw_data: pd.DataFrame) -> pd.DataFrame: """Remove null values and duplicates. :param raw_data: Raw data from previous step :return: Cleaned DataFrame """ pass def rolling_7d(cleaned_data: pd.DataFrame) -> pd.Series: """Calculate 7-day rolling average of sales. :param cleaned_data: Cleaned data :return: 7-day rolling average """ pass def top_10(rolling_7d: pd.Series) -> pd.Series: """Get top 10 days by value. :param rolling_7d: Rolling average values :return: Top 10 values """ pass ``` **Type Annotation Strategy:** - **Known types**: Use specific types (pd.DataFrame, pd.Series, int, str) - **Uncertain**: Use `Any` - you'll fix this later with MonkeyType - **Complex types**: Start with `Any`, refine incrementally ### Step 3: Validate DAG with Hamilton CLI **Verify the graph structure before implementing logic** Use the Hamilton CLI to validate: ```bash # Validate the DAG structure hamilton build --module my_pipeline # Visualize the graph hamilton build --module my_pipeline --output dag.png # List all nodes hamilton build --module my_pipeline --list-nodes ``` **What to check:** - ✅ No circular dependencies - ✅ All edges connect properly - ✅ Input nodes are identified correctly - ✅ Output nodes are what you expect - ✅ No typos in function/parameter names **If validation fails:** - Fix the function signatures (parameter names, function names) - Re-run `hamilton build` - Don't proceed to implementation until validation passes ### Step 4: Function Signatures → TDD Implementation **Implement nodes incrementally with test-driven development** This is where the workflow shines for LLMs. Instead of writing all code at once: **Process:** 1. Pick ONE node to implement (start with inputs, work forward) 2. Write a test for that node first 3. Implement the node to pass the test 4. Run the test to verify 5. Move to the next node **Example - Node by node:** ```python # Test for raw_data def test_raw_data(tmp_path): """Test CSV loading.""" # Setup csv_file = tmp_path / "data.csv" csv_file.write_text("col_a,col_b\n1,2\n3,4") # Execute result = raw_data(str(csv_file)) # Verify assert isinstance(result, pd.DataFrame) assert len(result) == 2 assert list(result.columns) == ["col_a", "col_b"] # Implementation def raw_data(data_path: str) -> pd.DataFrame: """Load raw data from CSV file.""" return pd.read_csv(data_path) ``` ```python # Test for cleaned_data def test_cleaned_data(): """Test data cleaning.""" # Setup raw = pd.DataFrame({"col_a": [1, 2, None, 4], "col_b": [1, 1, 2, 2]}) # Execute result = cleaned_data(raw) # Verify assert len(result) == 3 # One null removed assert result["col_a"].isna().sum() == 0 # Implementation def cleaned_data(raw_data: pd.DataFrame) -> pd.DataFrame: """Remove null values and duplicates.""" return raw_data.dropna().drop_duplicates() ``` **Why this works:** - **Focused context**: LLM only thinks about one node at a time - **Immediate validation**: Each node is tested before moving on - **Easy refactoring**: Well-tested nodes can be safely refactored - **Handles complexity**: Build complex pipelines without "spaghetti code" - **Clear progress**: Know exactly what's done and what's left ### Step 5: Handle Type Annotations Pragmatically **Use MonkeyType or runtime inference to add accurate types** Don't spend time guessing types upfront. Instead: **Option A: Use `Any` everywhere initially** ```python from typing import Any def my_function(input_data: Any) -> Any: """Process data.""" # Implementation here pass ``` **Option B: Disable type validation in Driver** ```python from hamilton import driver from hamilton.lifecycle import base # Use an adapter that doesn't validate types class NoTypeCheckAdapter(base.BaseDo): """Adapter that skips type validation.""" pass dr = driver.Driver({}, module, adapter=NoTypeCheckAdapter()) ``` **Option C: Run MonkeyType to infer types** After implementing and testing: ```bash # Record runtime types monkeytype run -m pytest tests/ # Generate type stubs monkeytype stub my_pipeline # Apply types to source monkeytype apply my_pipeline ``` MonkeyType will: - Observe actual values passed at runtime - Infer concrete types from observations - Generate accurate type annotations - Update your source files **Result**: Accurate types without upfront guessing! **Important**: MonkeyType isn't perfect, but it's: - ✅ Much faster than manual annotation - ✅ Based on real runtime behavior - ✅ Good enough for first drafts - ✅ Easy to refine manually later ## Complete Example **User Request:** "Create a pipeline that loads JSON data, filters by status='active', enriches with external API data, and outputs to Parquet" **Step 1: DOT Graph** ```dot digraph pipeline { // Inputs json_path [shape=box, color=blue] api_endpoint [shape=box, color=blue] output_path [shape=box, color=blue] // Pipeline (arrows show data flow: upstream -> downstream) json_path -> raw_data raw_data -> active_records active_records -> enriched_data api_endpoint -> enriched_data enriched_data -> final_output output_path -> final_output } ``` **Step 2: Function Signatures** ```python """Data enrichment pipeline.""" from typing import Any import pandas as pd def raw_data(json_path: str) -> pd.DataFrame: """Load data from JSON file. :param json_path: Path to JSON file :return: Raw DataFrame """ pass def active_records(raw_data: pd.DataFrame) -> pd.DataFrame: """Filter for active status only. :param raw_data: Raw data :return: Filtered DataFrame """ pass def enriched_data(active_records: pd.DataFrame, api_endpoint: str) -> pd.DataFrame: """Enrich with external API data. :param active_records: Active records :param api_endpoint: API endpoint URL :return: Enriched DataFrame """ pass def final_output(enriched_data: pd.DataFrame, output_path: str) -> str: """Save to Parquet file. :param enriched_data: Enriched data :param output_path: Output file path :return: Path to saved file """ pass ``` **Step 3: Validate** ```bash hamilton build --module enrichment_pipeline # Output: ✓ DAG is valid, 4 nodes, 0 cycles ``` **Step 4: TDD Implementation** *Test 1:* ```python def test_raw_data(tmp_path): json_file = tmp_path / "data.json" json_file.write_text('[{"id": 1, "status": "active"}]') result = raw_data(str(json_file)) assert len(result) == 1 ``` *Implementation 1:* ```python def raw_data(json_path: str) -> pd.DataFrame: return pd.read_json(json_path) ``` *Test 2:* ```python def test_active_records(): raw = pd.DataFrame([{"id": 1, "status": "active"}, {"id": 2, "status": "inactive"}]) result = active_records(raw) assert len(result) == 1 assert result.iloc[0]["status"] == "active" ``` *Implementation 2:* ```python def active_records(raw_data: pd.DataFrame) -> pd.DataFrame: return raw_data[raw_data["status"] == "active"] ``` ...continue for each node... **Step 5: Add Types with MonkeyType** ```bash monkeytype run -m pytest tests/ monkeytype apply enrichment_pipeline ``` Result: Accurate type annotations added automatically! ## Key Benefits of This Workflow ### For LLMs (Claude): - **Structured thinking**: Each step has clear boundaries - **Token efficient**: DOT is concise, avoids redundant code - **Incremental**: Build one piece at a time - **Validatable**: Check work at each step - **Manageable complexity**: Never overwhelmed with full implementation ### For Developers: - **Visual validation**: See the DAG before coding - **Test coverage**: Every node is tested - **Maintainable**: Well-structured, documented code - **Debuggable**: Issues isolated to specific nodes - **Refactorable**: Safe to modify with test coverage ## Advanced Patterns ### Using Function Modifiers Add decorators in Step 2: ```python from hamilton.function_modifiers import parameterize @parameterize( rolling_7d={"window": 7}, rolling_30d={"window": 30}, rolling_90d={"window": 90}, ) def rolling_average(data: pd.Series, window: int) -> pd.Series: """Calculate rolling average. :param data: Input data :param window: Rolling window size :return: Rolling average """ pass ``` Update DOT graph to reflect expanded nodes: ```dot data -> rolling_7d data -> rolling_30d data -> rolling_90d ``` ### Async Functions Mark async in Step 2: ```python async def api_data(endpoint: str, record_id: int) -> dict: """Fetch data from API. :param endpoint: API endpoint :param record_id: Record ID to fetch :return: API response data """ pass ``` Use async driver: ```python from hamilton import async_driver dr = await async_driver.Builder().with_modules(module).build() ``` ## Common Pitfalls ### ❌ Implementing before validating Don't write function bodies before running `hamilton build`. Validate structure first! ### ❌ Writing all code at once Don't implement all nodes before testing. Go node-by-node with TDD. ### ❌ Guessing type annotations Don't waste time guessing types. Use `Any` + MonkeyType or disable validation initially. ### ❌ Skipping the DOT step Don't jump straight to code. The DOT graph is your blueprint - validate it first. ### ❌ Not running tests Don't implement the next node until the current one passes its test. ## Quick Reference ```bash # Step 3: Validate DAG hamilton build --module my_module hamilton build --module my_module --output dag.png # Step 4: Run tests pytest tests/test_my_module.py::test_my_function -v # Step 5: Infer types monkeytype run -m pytest tests/ monkeytype apply my_module # Full pipeline validation python -c "from hamilton import driver; import my_module; driver.Driver({}, my_module)" ``` ## Summary 1. **DOT first**: Design the graph in DOT language (token-efficient, validatable) 2. **Signatures next**: Convert to function signatures with docstrings 3. **Validate early**: Use `hamilton build` before implementing 4. **TDD everything**: Implement one node at a time with tests 5. **Types pragmatically**: Use Any + MonkeyType or disable validation initially **This workflow enables Claude to build complex Hamilton DAGs systematically, avoiding the pitfalls of monolithic implementation.**