Agent task decomposition cost efficiency: how breaking work into subtasks changes your LLM API bill

import anthropic
from runguard import guard, BudgetExceededError
from dataclasses import dataclass
from typing import Any, Optional

client = anthropic.Anthropic()

# Per-step budget caps: allocate your total task budget across steps
# Step allocations should reflect expected complexity, not equal splits.
@dataclass
class PipelineStep:
    name: str
    system_prompt: str
    budget_usd: float           # hard cap for this step
    context_keys: list[str]     # which prior step outputs to include as context

PIPELINE = [
    PipelineStep(
        name="understand_requirements",
        system_prompt="Extract structured requirements from the user's request. Output JSON.",
        budget_usd=0.05,
        context_keys=[],        # no prior context needed — just the original request
    ),
    PipelineStep(
        name="write_tests",
        system_prompt="Write unit tests for the requirements. Focus on edge cases.",
        budget_usd=0.15,
        context_keys=["understand_requirements"],   # only requirements, not original request verbatim
    ),
    PipelineStep(
        name="write_implementation",
        system_prompt="Implement code that passes the provided tests.",
        budget_usd=0.20,
        context_keys=["understand_requirements", "write_tests"],
    ),
    PipelineStep(
        name="review",
        system_prompt="Review the implementation for correctness and completeness against the tests.",
        budget_usd=0.10,
        context_keys=["write_tests", "write_implementation"],  # NOT requirements — irrelevant to review
    ),
]

def build_step_messages(
    original_request: str,
    step: PipelineStep,
    prior_outputs: dict[str, str],
) -> list[dict]:
    """Build a message list with only the context slices this step needs."""
    content_parts = [f"Original request:\n{original_request}"]
    for key in step.context_keys:
        if key in prior_outputs:
            content_parts.append(f"\n{key.replace('_', ' ').title()} output:\n{prior_outputs[key]}")
    return [{"role": "user", "content": "\n\n".join(content_parts)}]

def run_pipeline(user_request: str) -> dict[str, Any]:
    """Run all pipeline steps, respecting per-step budget caps."""
    prior_outputs: dict[str, str] = {}
    total_spent = 0.0

    for step in PIPELINE:
        messages = build_step_messages(user_request, step, prior_outputs)

        # Create a fresh guard per step with the step's budget cap
        @guard(budget_usd=step.budget_usd, loop_max_repeats=3)
        def _run_step(msgs: list, sys: str) -> str:
            resp = client.messages.create(
                model="claude-haiku-4-5-20251001",
                max_tokens=1024,
                system=sys,
                messages=msgs,
            )
            return resp.content[0].text

        try:
            output = _run_step(messages, step.system_prompt)
            prior_outputs[step.name] = output
            print(f"  [pipeline] {step.name}: OK ({len(output)} chars)")
        except BudgetExceededError as e:
            print(f"  [pipeline] {step.name}: BUDGET EXCEEDED — {e}")
            # Abort pipeline; return partial results
            return {
                "status": "budget_exceeded",
                "failed_step": step.name,
                "partial_outputs": prior_outputs,
            }

    return {"status": "ok", "outputs": prior_outputs}


# Example: batch extraction to avoid re-sending the same large context 10 times
def batch_extract_fields(document: str, fields: list[str]) -> dict[str, str]:
    """Extract multiple fields in one call instead of N separate calls."""
    field_list = "\n".join(f"- {f}" for f in fields)
    prompt = f"""Extract the following fields from the document. Return as JSON.

Fields to extract:
{field_list}

Document:
{document}"""

    @guard(budget_usd=0.20, loop_max_repeats=3)
    def _extract(p: str) -> str:
        resp = client.messages.create(
            model="claude-haiku-4-5-20251001",
            max_tokens=512,
            messages=[{"role": "user", "content": p}],
        )
        return resp.content[0].text

    return _extract(prompt)  # one call, one context-load, all fields