# How to Optimize Skill Performance for Complex Tasks in Anthropic Skills

> Optimize skill performance for complex Anthropic Skills tasks. Decompose work, declare dependencies, use caching, and configure AsyncExecutor for maximum concurrency.

- Repository: [Anthropic/skills](https://github.com/anthropics/skills)
- Tags: performance
- Published: 2026-02-16

---

**To optimize skill performance for complex tasks, decompose work into independent sub-skills, declare explicit dependencies to enable DAG-based parallel execution, leverage the built-in cache decorator for deterministic operations, and configure the AsyncExecutor to respect rate limits while maximizing concurrency.**

The `anthropics/skills` repository provides a lightweight framework for exposing reusable functionality to Claude. When user requests involve many inter-dependent operations—such as data extraction, transformation, and external API calls—optimizing skill performance for complex tasks becomes critical to reducing latency and token costs.

## Architectural Foundation for High-Performance Skills

Understanding the core components is essential before implementing optimizations.

### Core Components

The framework's architecture is defined across several key files:

| Component | Responsibility | File Path |
|-----------|----------------|-----------|
| **Skill Definition** | Declares input/output schemas and the `run` implementation. | [`skills/skill.py`](https://github.com/anthropics/skills/blob/main/skills/skill.py) |
| **Skill Registry** | Global mapping of skill names to classes for function-calling lookup. | [`skills/registry.py`](https://github.com/anthropics/skills/blob/main/skills/registry.py) |
| **Skill Runner** | Orchestrates execution, validates payloads, resolves dependencies, and handles caching. | [`skills/runner.py`](https://github.com/anthropics/skills/blob/main/skills/runner.py) |
| **Cache Layer** | Memoizes deterministic results to avoid repeated work. | [`skills/cache.py`](https://github.com/anthropics/skills/blob/main/skills/cache.py) |
| **Concurrency Helper** | Wraps `asyncio` to run independent skills in parallel with rate-limit awareness. | [`skills/executor.py`](https://github.com/anthropics/skills/blob/main/skills/executor.py) |
| **Prompt Utils** | Generates concise function signatures to minimize token usage. | [`skills/prompt_utils.py`](https://github.com/anthropics/skills/blob/main/skills/prompt_utils.py) |

### Execution Flow and DAG Construction

When processing complex requests, the `SkillRunner` in [`skills/runner.py`](https://github.com/anthropics/skills/blob/main/skills/runner.py) builds a directed acyclic graph (DAG) of required skills based on declared dependencies. The execution flow follows these stages:

1. **Planning**: The runner identifies all required skills and constructs the dependency graph.
2. **Parallelization**: Leaf nodes with no inter-dependencies are dispatched to the `AsyncExecutor` simultaneously.
3. **Cache Lookup**: Before invocation, the runner checks `SkillCache`. If a cached response exists (keyed by deterministic input hash), the skill is skipped.
4. **Aggregation**: Outputs from completed leaf skills feed into parent nodes until the root skill returns the final result.

This architecture transforms potentially linear *O(n)* execution into *O(depth of DAG)*, significantly optimizing skill performance for complex tasks.

## Optimization Techniques for Complex Workflows

### Fine-Grained Skill Decomposition

Split monolithic skills into reusable, independent sub-skills. This allows the `SkillRunner` to identify parallelization opportunities. Instead of a single skill handling "extract, transform, and load," create three separate skills that can be chained or run concurrently where dependencies permit.

### Dependency Declaration and DAG Parallelization

Explicitly declare dependencies using the `depends_on` attribute in your `Skill` subclass. This enables the `SkillRunner` to build an accurate DAG and maximize parallel execution. Without explicit declarations, the framework defaults to sequential execution, missing optimization opportunities.

### Result Caching with the @cached Decorator

For deterministic operations—such as static lookups or pure transformations—apply the `@cached` decorator from [`skills/cache.py`](https://github.com/anthropics/skills/blob/main/skills/cache.py). This memoizes results based on input hashes, eliminating redundant API calls or computations across multiple invocations.

### Async Execution and Concurrency Control

Leverage `asyncio` through the `AsyncExecutor` in [`skills/executor.py`](https://github.com/anthropics/skills/blob/main/skills/executor.py). Configure `max_concurrency` per skill to respect external API rate limits while maximizing throughput. The executor automatically manages the event loop and backpressure for you.

### Prompt Length Minimization

Use `prompt_utils.shrink_schema` from [`skills/prompt_utils.py`](https://github.com/anthropics/skills/blob/main/skills/prompt_utils.py) to trim large JSON schemas before sending them to Claude. Concise function signatures reduce token consumption and improve response times, particularly when many skills are registered simultaneously.

## Implementation Examples

### Declaring a Parallel-Friendly Skill

The following example from [`skills/translate.py`](https://github.com/anthropics/skills/blob/main/skills/translate.py) demonstrates a skill designed for concurrent execution with caching enabled:

```python

# file: skills/translate.py

from skills.skill import Skill
from skills.cache import cached

class TranslateSkill(Skill):
    name = "translate"
    input_schema = {"text": "string", "target_lang": "string"}
    output_schema = {"translated_text": "string"}

    @cached  # <-- memoises deterministic translations (optional)

    async def run(self, text: str, target_lang: str) -> dict:
        # Example: call an async translation API

        result = await self.deps.http_client.post(
            "https://api.example.com/translate",
            json={"q": text, "target": target_lang},
        )
        return {"translated_text": result["translation"]}

```

### Building a DAG for Complex Workflows

This example from [`examples/complex_task.py`](https://github.com/anthropics/skills/blob/main/examples/complex_task.py) shows how to construct a dependency graph for multi-step processing:

```python

# file: examples/complex_task.py

from skills.registry import SkillRegistry
from skills.runner import SkillRunner
from skills.translate import TranslateSkill
from skills.summarize import SummarizeSkill

# Register the skills (normally auto-discovered)

SkillRegistry.register(TranslateSkill)
SkillRegistry.register(SummarizeSkill)

async def main():
    runner = SkillRunner()

    # Define a high-level plan: translate → summarize → sentiment analysis

    plan = {
        "summarize": {
            "skill": "summarize",
            "depends_on": ["translate"],
            "input": {"text": "$translate.translated_text"},
        },
        "translate": {
            "skill": "translate",
            "input": {"text": "Long English article …", "target_lang": "fr"},
        },
    }

    result = await runner.execute(plan)
    print("Final summary (French):", result["summarize"]["summary"])

```

The runner automatically resolves that `summarize` depends on `translate`, executes them in the correct order, and could run additional independent skills in parallel if present.

### Enabling Caching Across Runs

The caching implementation in [`skills/cache.py`](https://github.com/anthropics/skills/blob/main/skills/cache.py) uses input hashing to memoize results:

```python

# file: skills/cache.py

from functools import wraps
import hashlib
import json
from typing import Any

_CACHE = {}

def _hash_inputs(*args, **kwargs) -> str:
    payload = json.dumps([args, kwargs], sort_keys=True).encode()
    return hashlib.sha256(payload).hexdigest()

def cached(fn):
    @wraps(fn)
    async def wrapper(*args, **kwargs):
        key = f"{fn.__qualname__}:{_hash_inputs(*args, **kwargs)}"
        if key in _CACHE:
            return _CACHE[key]
        result = await fn(*args, **kwargs)
        _CACHE[key] = result
        return result
    return wrapper

```

Subsequent executions with identical inputs retrieve results from `_CACHE` instantly, eliminating redundant computation or API calls.

## Key Files and Their Roles

| Path | Purpose | Direct Link |
|------|---------|-------------|
| [`skills/skill.py`](https://github.com/anthropics/skills/blob/main/skills/skill.py) | Base class for all skills; defines schema handling and the `run` contract. | [skills/skill.py](https://github.com/anthropics/skills/blob/main/skills/skill.py) |
| [`skills/registry.py`](https://github.com/anthropics/skills/blob/main/skills/registry.py) | Global registry mapping skill names to implementations; used by Claude's function-calling layer. | [skills/registry.py](https://github.com/anthropics/skills/blob/main/skills/registry.py) |
| [`skills/runner.py`](https://github.com/anthropics/skills/blob/main/skills/runner.py) | Core orchestrator that builds DAGs, manages async execution, and handles caching. | [skills/runner.py](https://github.com/anthropics/skills/blob/main/skills/runner.py) |
| [`skills/executor.py`](https://github.com/anthropics/skills/blob/main/skills/executor.py) | Concurrency helper wrapping `asyncio` with rate-limit awareness. | [skills/executor.py](https://github.com/anthropics/skills/blob/main/skills/executor.py) |
| [`skills/cache.py`](https://github.com/anthropics/skills/blob/main/skills/cache.py) | Memoization utilities for deterministic skill results. | [skills/cache.py](https://github.com/anthropics/skills/blob/main/skills/cache.py) |
| [`skills/prompt_utils.py`](https://github.com/anthropics/skills/blob/main/skills/prompt_utils.py) | Functions to minimize token usage by shrinking JSON schemas. | [skills/prompt_utils.py](https://github.com/anthropics/skills/blob/main/skills/prompt_utils.py) |
| [`examples/complex_task.py`](https://github.com/anthropics/skills/blob/main/examples/complex_task.py) | End-to-end demonstration of multi-skill pipelines with parallelization. | [examples/complex_task.py](https://github.com/anthropics/skills/blob/main/examples/complex_task.py) |

## Summary

- **Decompose monolithic skills** into smaller, independent sub-skills to maximize parallelization opportunities through the DAG builder in [`skills/runner.py`](https://github.com/anthropics/skills/blob/main/skills/runner.py).
- **Declare explicit dependencies** using the `depends_on` attribute so the `SkillRunner` can identify which skills run in parallel versus sequentially.
- **Apply the `@cached` decorator** from [`skills/cache.py`](https://github.com/anthropics/skills/blob/main/skills/cache.py) to deterministic operations, eliminating redundant API calls and computations across multiple invocations.
- **Leverage `AsyncExecutor`** in [`skills/executor.py`](https://github.com/anthropics/skills/blob/main/skills/executor.py) to run independent skills concurrently while respecting external rate limits through configurable `max_concurrency` settings.
- **Minimize token usage** by using `prompt_utils.shrink_schema` to generate concise function signatures, reducing context window pressure when many skills are registered.

## Frequently Asked Questions

### How does the SkillRunner determine which skills can run in parallel?

The `SkillRunner` in [`skills/runner.py`](https://github.com/anthropics/skills/blob/main/skills/runner.py) constructs a directed acyclic graph (DAG) by analyzing the `depends_on` declarations in each skill's configuration. Skills that appear as leaf nodes with no dependencies, or those whose dependencies have already been satisfied, are dispatched simultaneously to the `AsyncExecutor`. This transforms sequential *O(n)* execution into *O(depth of DAG)* wall-clock time.

### What types of operations benefit most from the @cached decorator?

Deterministic operations—those that produce identical outputs given identical inputs—benefit most from caching. Examples include static data lookups, mathematical transformations, translation of fixed text passages, and embedding generation for unchanging documents. The `@cached` decorator in [`skills/cache.py`](https://github.com/anthropics/skills/blob/main/skills/cache.py) hashes input arguments using SHA-256, storing results in memory (or an external store if configured) to eliminate redundant computation or API calls.

### How can I prevent rate limiting when running many skills concurrently?

Configure the `max_concurrency` parameter in [`skills/executor.py`](https://github.com/anthropics/skills/blob/main/skills/executor.py) on a per-skill basis. The `AsyncExecutor` maintains separate semaphore pools for each skill type, ensuring that high-throughput operations (like batch embedding requests) do not overwhelm external APIs while still maximizing parallelism for other independent skills. Additionally, implement exponential backoff within individual skill `run` methods for transient failures.

### Is it possible to stream partial results for long-running skills?

Yes. For skills processing large outputs (such as long document generation or multi-page data extraction), implement a generator in your `Skill.run` method and wrap it with `SkillRunner.stream`. This approach keeps Claude's context window small by yielding incremental results rather than buffering the entire output, reducing memory pressure and improving perceived responsiveness for end users.