Part 3: Enhanced Implementation Flow

In the previous parts, we built a single-shot implementation flow and a planning flow. Now we'll enhance our implementation flow to work with the tasks generated by the planner.

This enhanced implementation flow will:

1. Accept a task specification from the planner
2. Dynamically analyze the required context for that specific task
3. Generate and execute precise file operations
4. Provide better error handling and validation

Creating the Enhanced Implementation Flow

Let's create an enhanced implementation flow that can work with tasks from the planner:

# src/agents/planner_coder/flows/implementation.py
from dhenara.agent.dsl import (
    PLACEHOLDER,
    AIModelNode,
    AIModelNodeSettings,
    EventType,
    FileOperationNode,
    FileOperationNodeSettings,
    FlowDefinition,
    FolderAnalyzerNode,
    FolderAnalyzerSettings,
)
from dhenara.ai.types import (
    AIModelCallConfig,
    ObjectTemplate,
    Prompt,
)

from ..types import TaskImplementation

# Constants
global_data_directory = "$expr{run_root}/global_data"
models = [
    "claude-3-7-sonnet",
    "gpt-4.1",
    "gpt-4.1-mini",
]
test_mode = False

# Create the implementation flow
implementation_flow = FlowDefinition().vars(
    {
        "task_spec": PLACEHOLDER,  # This will be filled by the coordinator
    }
)

# 1. Dynamic Folder Analysis based on task requirements
implementation_flow.node(
    "dynamic_context_analysis",
    FolderAnalyzerNode(
        settings=FolderAnalyzerSettings(
            base_directory=global_data_directory,
            operations_template=ObjectTemplate(expression="$expr{task_spec.required_context}"),
        ),
    ),
)

# 2. Code Generation Node
implementation_flow.node(
    "code_generator",
    AIModelNode(
        pre_events=[EventType.node_input_required],
        settings=AIModelNodeSettings(
            models=models,
            system_instructions=[
                "You are a professional code implementation agent specialized in executing precise file operations.",
                "Your task is to generate the exact file operations necessary to implement requested code changes - nothing more, nothing less.",
                "Generate machine-executable operations that require zero human intervention.",
                "ALLOWED OPERATIONS:",
                "- create_file(file_path, content)",
                "- delete_file(file_path)",
                "- create_directory(directory_path)",
                "- move_file(source_path, destination_path)",
                "IMPLEMENTATION GUIDELINES:",
                "1. For complete file replacement, use delete_file followed by create_file.",
                "2. Maintain the project's existing code style, indentation, and formatting conventions.",
                "3. Use Python 3.10 style in all code examples.",
                "4. Focus on implementing ONLY the specific task assigned to you, not the entire project.",
                "5. Reference the analyzed context to understand the existing code structure.",
            ],
            prompt=Prompt.with_dad_text(
                text=(
                    "## Task Specification\n"
                    "Task ID: $expr{task_spec.task_id}\n"
                    "Order: $expr{task_spec.order}\n"
                    "Description: $expr{task_spec.description}\n\n"
                    "## Repository Context\n"
                    "$expr{$hier{dynamic_context_analysis}.outcome.results}\n\n"
                    "## Implementation Requirements\n"
                    "1. Generate precise file operations that can be executed programmatically\n"
                    "2. Focus only on implementing THIS specific task, not the entire project\n"
                    "3. Consider the relevant context when making implementation decisions\n"
                    "4. Follow a complete implementation strategy that addresses all aspects of this specific task\n\n"
                    "## Output Format\n"
                    "Return a TaskImplementation object with the task_id set to: $expr{task_spec.task_id}\n"
                ),
            ),
            model_call_config=AIModelCallConfig(
                structured_output=TaskImplementation,
                test_mode=test_mode,
                max_output_tokens=16000,  # Increased token limit for complex implementations
                reasoning=True,  # Enable reasoning for better transparency
            ),
        ),
    ),
)

# 3. File Operation Node
implementation_flow.node(
    "file_operations",
    FileOperationNode(
        settings=FileOperationNodeSettings(
            base_directory=global_data_directory,
            operations_template=ObjectTemplate(
                expression="$expr{ $hier{code_generator}.outcome.structured.file_operations }",
            ),
            stage=True,  # Stage changes without committing
            commit=False,  # Commit handled in the coordinator
            commit_message="Implemented task: $expr{task_spec.task_id}",
        ),
    ),
)

Updating the Handler

Now we need to update our handler to support the enhanced implementation flow:

# src/agents/planner_coder/handler.py
# (Add to existing handler.py file)

async def planner_coder_input_handler(event: NodeInputRequiredEvent):
    """Handles input required events for the planner coder"""
    if event.node_type == FlowNodeTypeEnum.ai_model_call:
        node_input = None

        if event.node_id == "plan_generator":
            # ... existing plan_generator handler code ...

        elif event.node_id == "code_generator":
            node_input = await get_ai_model_node_input(
                node_def_settings=event.node_def_settings,
                models=models,
            )
            # No additional prompt variables needed here
            # The task_spec is already passed to the flow as a variable

        else:
            print(f"WARNING: Unhandled ai_model_call input event for node {event.node_id}")

        event.input = node_input
        event.handled = True

    elif event.node_type == FlowNodeTypeEnum.folder_analyzer:
        # ... existing folder_analyzer handler code ...

Testing the Enhanced Implementation Flow

To test just the implementation flow (without the coordinator), we can create a simple test agent that uses a predefined task specification:

# src/agents/planner_coder/test_implementation.py
from dhenara.agent.dsl import AgentDefinition, FlowDefinition
from dhenara.agent.dsl.inbuilt.flow_nodes.defs.types import FolderAnalysisOperation

from .flows.implementation import implementation_flow
from .types import TaskSpec

# Create a test task specification
test_task = TaskSpec(
    order=1,
    task_id="create_flask_app",
    description="Create a basic Flask application with a main.py file that sets up a simple web server with a 'Hello, World!' route at '/'.",
    required_context=[
        FolderAnalysisOperation(
            operation_type="analyze_folder",
            path=".",
            max_depth=2,
            content_read_mode="structure",
        ),
    ],
)

# Create a test flow that uses our implementation flow with the test task
test_flow = FlowDefinition()
test_flow.subflow(
    "implementation",
    implementation_flow,
    variables={
        "task_spec": test_task,
    },
)

# Create a test agent
test_agent = AgentDefinition()
test_agent.flow(
    "test_implementation",
    test_flow,
)

You can then create a runner for this test agent and run it to see how the implementation flow works with a specific task.

Key Improvements in the Enhanced Implementation Flow

Let's highlight the key improvements in our enhanced implementation flow compared to the single-shot implementation:

1. Task-specific context: The flow now analyzes only the specific files needed for the current task, rather than the entire codebase

2. Focus on specific requirements: The prompt emphasizes implementing only the current task, not the entire project

3. Better error handling: With reasoning enabled, we get more transparency into the implementation process

4. Increased token limit: We've increased the max_output_tokens to handle more complex implementations

5. Dynamic operation: The flow can work with any task specification provided by the planner

What's Next?

Now that we have both a planning flow and an enhanced implementation flow, we need to create a coordinator flow that can:

1. Run the planner to generate tasks
2. Iterate through each task in the plan
3. Run the implementation flow for each task
4. Handle errors and track progress

In Part 4: Coordinator Flow, we'll bring everything together to create our complete command-line coding assistant.