Core API

This page is auto-generated from docstrings via mkdocstrings. Source-of-truth lives in the Python files.

Patterns

consensus

executionkit.patterns.consensus.consensus async

consensus(provider: LLMProvider, prompt: str, *, num_samples: int = 5, strategy: VotingStrategy | str = 'majority', temperature: float = _DEFAULT_TEMPERATURE, max_tokens: int = DEFAULT_MAX_TOKENS, max_concurrency: int = _DEFAULT_CONSENSUS_CONCURRENCY, retry: RetryConfig | None = None, max_cost: TokenUsage | None = None, trace: TraceCallback | None = None) -> PatternResult[str]

Run parallel LLM samples and aggregate via voting.

Fires num_samples concurrent completions and applies the chosen voting strategy to determine the winning response.

Parameters:

Name	Type	Description	Default
provider	LLMProvider	LLM provider to call.	required
prompt	str	User prompt sent identically to every sample.	required
num_samples	int	Number of parallel completions to request. Must be >= 1.	5
strategy	VotingStrategy | str	"majority" (most common wins) or "unanimous" (all must agree). Accepts a :class:VotingStrategy enum or a plain string.	'majority'
temperature	float	Sampling temperature (higher = more diverse).	_DEFAULT_TEMPERATURE
max_tokens	int	Maximum tokens per completion.	DEFAULT_MAX_TOKENS
max_concurrency	int	Semaphore limit for parallel calls.	_DEFAULT_CONSENSUS_CONCURRENCY
retry	RetryConfig | None	Optional retry configuration per call.	None
max_cost	TokenUsage | None	Optional token/call budget. Passed to each individual checked_complete call. None means unlimited.	None
trace	TraceCallback | None	Optional structured trace callback.	None

Returns:

Name	Type	Description
A	PatternResult[str]	class:PatternResult whose value is the winning response,
	PatternResult[str]	score is the agreement ratio, and metadata includes
	PatternResult[str]	agreement_ratio, unique_responses, and tie_count.

Raises:

Type	Description
ConsensusFailedError	When strategy="unanimous" and responses are not all identical.
ValueError	If num_samples is less than 1.

Metadata

agreement_ratio (float): Fraction of samples matching the winner (0.0-1.0). unique_responses (int): Number of distinct response strings observed. tie_count (int): Number of responses that tied for the top vote count.

Source code in executionkit/patterns/consensus.py

async def consensus(
    provider: LLMProvider,
    prompt: str,
    *,
    num_samples: int = 5,
    strategy: VotingStrategy | str = "majority",
    temperature: float = _DEFAULT_TEMPERATURE,
    max_tokens: int = DEFAULT_MAX_TOKENS,
    max_concurrency: int = _DEFAULT_CONSENSUS_CONCURRENCY,
    retry: RetryConfig | None = None,
    # NOTE (F-03 verified): max_cost is implemented and forwarded to every
    # checked_complete() call below, enabling budget-aware pipe() chains.
    # See executionkit/compose.py _filter_kwargs() for propagation logic.
    max_cost: TokenUsage | None = None,
    trace: TraceCallback | None = None,
) -> PatternResult[str]:
    """Run parallel LLM samples and aggregate via voting.

    Fires ``num_samples`` concurrent completions and applies the chosen
    voting strategy to determine the winning response.

    Args:
        provider: LLM provider to call.
        prompt: User prompt sent identically to every sample.
        num_samples: Number of parallel completions to request. Must be >= 1.
        strategy: ``"majority"`` (most common wins) or ``"unanimous"``
            (all must agree).  Accepts a :class:`VotingStrategy` enum or
            a plain string.
        temperature: Sampling temperature (higher = more diverse).
        max_tokens: Maximum tokens per completion.
        max_concurrency: Semaphore limit for parallel calls.
        retry: Optional retry configuration per call.
        max_cost: Optional token/call budget. Passed to each individual
            ``checked_complete`` call. ``None`` means unlimited.
        trace: Optional structured trace callback.

    Returns:
        A :class:`PatternResult` whose ``value`` is the winning response,
        ``score`` is the agreement ratio, and ``metadata`` includes
        ``agreement_ratio``, ``unique_responses``, and ``tie_count``.

    Raises:
        ConsensusFailedError: When ``strategy="unanimous"`` and responses
            are not all identical.
        ValueError: If ``num_samples`` is less than 1.

    Metadata:
        agreement_ratio (float): Fraction of samples matching the winner (0.0-1.0).
        unique_responses (int): Number of distinct response strings observed.
        tie_count (int): Number of responses that tied for the top vote count.
    """
    if num_samples < 1:
        raise ValueError(f"num_samples must be >= 1, got {num_samples}")
    if max_concurrency < 1:
        raise ValueError(f"max_concurrency must be >= 1, got {max_concurrency}")
    if max_tokens < 1:
        raise ValueError(f"max_tokens must be >= 1, got {max_tokens}")

    if isinstance(strategy, str):
        strategy = VotingStrategy(strategy)

    tracker = CostTracker()
    messages: list[dict[str, Any]] = [user_message(prompt)]

    coros = [
        checked_complete(
            provider,
            messages,
            tracker,
            budget=max_cost,
            retry=retry,
            trace=trace,
            temperature=temperature,
            max_tokens=max_tokens,
        )
        for _ in range(num_samples)
    ]

    responses = await gather_strict(coros, max_concurrency=max_concurrency)
    contents = [r.content for r in responses]

    # Voting semantics live in engine/voting.py, shared verbatim with the
    # Message Batches fan-out (executionkit/batches.py).
    tally = tally_votes(contents, strategy)

    return PatternResult[str](
        value=tally.winner,
        score=tally.agreement_ratio,
        cost=tracker.to_usage(),
        metadata=MappingProxyType(
            {
                "agreement_ratio": tally.agreement_ratio,
                "unique_responses": tally.unique_responses,
                "tie_count": tally.tie_count,
            }
        ),
    )

refine_loop

executionkit.patterns.refine_loop.refine_loop async

refine_loop(provider: LLMProvider, prompt: str, *, evaluator: Evaluator | None = None, max_eval_chars: int = _DEFAULT_MAX_EVAL_CHARS, target_score: float = 0.9, max_iterations: int = 5, patience: int = 3, delta_threshold: float = 0.01, temperature: float = _DEFAULT_TEMPERATURE, max_tokens: int = DEFAULT_MAX_TOKENS, max_cost: TokenUsage | None = None, retry: RetryConfig | None = None, trace: TraceCallback | None = None, on_checkpoint: CheckpointCallback | None = None) -> PatternResult[str]

Iteratively refine an LLM response until convergence or budget exhaustion.

Generates an initial response, evaluates it, then enters a refinement loop. Each iteration asks the LLM to improve upon the previous output given its score. The loop terminates when the :class:ConvergenceDetector signals convergence (target score reached or score deltas stall beyond patience) or max_iterations is hit.

Parameters:

Name	Type	Description	Default
provider	LLMProvider	LLM provider to call.	required
prompt	str	The original user prompt.	required
evaluator	Evaluator | None	Async callable (text, provider) -> float returning a score in [0.0, 1.0]. If None, a default LLM-based evaluator scoring 0-10 (normalized to 0-1) is used.	None
target_score	float	Convergence target in [0.0, 1.0].	0.9
max_iterations	int	Maximum refinement iterations (excluding the initial generation).	5
patience	int	Stale-delta iterations before convergence is declared.	3
delta_threshold	float	Minimum meaningful score improvement.	0.01
temperature	float	Sampling temperature for generation calls.	_DEFAULT_TEMPERATURE
max_tokens	int	Maximum tokens per completion.	DEFAULT_MAX_TOKENS
max_cost	TokenUsage | None	Optional token/call budget.	None
retry	RetryConfig | None	Optional retry configuration per call.	None
trace	TraceCallback | None	Optional structured trace callback.	None
on_checkpoint	CheckpointCallback | None	Optional callback invoked after scoring each candidate (the initial generation as iteration 0, then each refinement iteration), receiving (iteration, state) where state is a JSON-serializable dict with keys iteration, current_text, current_score, and cost. May be sync or async; exceptions are logged and swallowed so a failing checkpoint never aborts the loop.	None

Returns:

Name	Type	Description
A	PatternResult[str]	class:PatternResult whose value is the best response seen,
	PatternResult[str]	score is its evaluation score, and metadata includes
	PatternResult[str]	iterations, converged, and score_history.

Metadata

iterations (int): Refinement iterations performed (0 = converged on first attempt). converged (bool): Whether the loop converged before max_iterations. score_history (list[float]): Score at each iteration including initial generation.

Source code in executionkit/patterns/refine_loop.py

async def refine_loop(
    provider: LLMProvider,
    prompt: str,
    *,
    evaluator: Evaluator | None = None,
    max_eval_chars: int = _DEFAULT_MAX_EVAL_CHARS,
    target_score: float = 0.9,
    max_iterations: int = 5,
    patience: int = 3,
    delta_threshold: float = 0.01,
    temperature: float = _DEFAULT_TEMPERATURE,
    max_tokens: int = DEFAULT_MAX_TOKENS,
    max_cost: TokenUsage | None = None,
    retry: RetryConfig | None = None,
    trace: TraceCallback | None = None,
    on_checkpoint: CheckpointCallback | None = None,
) -> PatternResult[str]:
    """Iteratively refine an LLM response until convergence or budget exhaustion.

    Generates an initial response, evaluates it, then enters a refinement
    loop.  Each iteration asks the LLM to improve upon the previous output
    given its score.  The loop terminates when the
    :class:`ConvergenceDetector` signals convergence (target score reached
    or score deltas stall beyond patience) or ``max_iterations`` is hit.

    Args:
        provider: LLM provider to call.
        prompt: The original user prompt.
        evaluator: Async callable ``(text, provider) -> float`` returning a
            score in ``[0.0, 1.0]``.  If ``None``, a default LLM-based
            evaluator scoring 0-10 (normalized to 0-1) is used.
        target_score: Convergence target in ``[0.0, 1.0]``.
        max_iterations: Maximum refinement iterations (excluding the
            initial generation).
        patience: Stale-delta iterations before convergence is declared.
        delta_threshold: Minimum meaningful score improvement.
        temperature: Sampling temperature for generation calls.
        max_tokens: Maximum tokens per completion.
        max_cost: Optional token/call budget.
        retry: Optional retry configuration per call.
        trace: Optional structured trace callback.
        on_checkpoint: Optional callback invoked after scoring each candidate
            (the initial generation as iteration 0, then each refinement
            iteration), receiving ``(iteration, state)`` where ``state`` is a
            JSON-serializable dict with keys ``iteration``, ``current_text``,
            ``current_score``, and ``cost``. May be sync or async; exceptions are
            logged and swallowed so a failing checkpoint never aborts the loop.

    Returns:
        A :class:`PatternResult` whose ``value`` is the best response seen,
        ``score`` is its evaluation score, and ``metadata`` includes
        ``iterations``, ``converged``, and ``score_history``.

    Metadata:
        iterations (int): Refinement iterations performed (0 = converged on
            first attempt).
        converged (bool): Whether the loop converged before ``max_iterations``.
        score_history (list[float]): Score at each iteration including initial
            generation.
    """
    _validate_refine_args(
        target_score,
        max_iterations,
        patience,
        delta_threshold,
        max_tokens,
        max_eval_chars,
    )

    tracker = CostTracker()
    convergence = ConvergenceDetector(
        delta_threshold=delta_threshold,
        patience=patience,
        score_threshold=target_score,
    )

    # Build default evaluator if none provided
    actual_evaluator: Evaluator = (
        evaluator
        if evaluator is not None
        else _make_default_evaluator(max_eval_chars, tracker, max_cost, retry, trace)
    )

    # Step 1: Generate initial response
    initial_messages: list[dict[str, Any]] = [{"role": "user", "content": prompt}]
    initial_response = await checked_complete(
        provider,
        initial_messages,
        tracker,
        max_cost,
        retry,
        trace,
        temperature=temperature,
        max_tokens=max_tokens,
    )

    best_text = initial_response.content
    best_score = await actual_evaluator(best_text, provider)
    score_history: list[float] = [best_score]
    if on_checkpoint is not None:
        await run_checkpoint(
            on_checkpoint,
            0,
            {
                "iteration": 0,
                "current_text": best_text,
                "current_score": best_score,
                "cost": dataclasses.asdict(tracker.snapshot()),
            },
            context="refine_loop",
        )
    converged = convergence.should_stop(best_score)
    iterations = 0

    # Step 2: Refinement loop
    if not converged:
        for iteration in range(1, max_iterations + 1):
            iterations = iteration

            refinement_messages: list[dict[str, Any]] = [
                {"role": "user", "content": prompt},
                {"role": "assistant", "content": best_text},
                {
                    "role": "user",
                    "content": (
                        f"The previous response scored {best_score:.2f} out of 1.0. "
                        "Please improve it. Focus on quality, completeness, and "
                        "accuracy. Provide the improved response only."
                    ),
                },
            ]

            refined_response = await checked_complete(
                provider,
                refinement_messages,
                tracker,
                max_cost,
                retry,
                trace,
                temperature=temperature,
                max_tokens=max_tokens,
            )

            refined_text = refined_response.content
            refined_score = await actual_evaluator(refined_text, provider)
            score_history.append(refined_score)

            # Track best result
            if refined_score > best_score:
                best_text = refined_text
                best_score = refined_score

            if on_checkpoint is not None:
                await run_checkpoint(
                    on_checkpoint,
                    iteration,
                    {
                        "iteration": iteration,
                        "current_text": refined_text,
                        "current_score": refined_score,
                        "cost": dataclasses.asdict(tracker.snapshot()),
                    },
                    context="refine_loop",
                )

            converged = convergence.should_stop(refined_score)
            if converged:
                break

    return PatternResult[str](
        value=best_text,
        score=best_score,
        cost=tracker.to_usage(),
        metadata=MappingProxyType(
            {
                "iterations": iterations,
                "converged": converged,
                "score_history": score_history,
            }
        ),
    )

react_loop

executionkit.patterns.react_loop.react_loop async

react_loop(provider: ToolCallingProvider, prompt: str | None = None, tools: Sequence[Tool] = (), *, messages: Sequence[dict[str, Any]] | None = None, max_rounds: int = _DEFAULT_MAX_ROUNDS, max_observation_chars: int = _DEFAULT_MAX_OBSERVATION_CHARS, tool_timeout: float | None = None, max_tool_calls_per_round: int = _DEFAULT_MAX_TOOL_CALLS_PER_ROUND, temperature: float = _DEFAULT_TEMPERATURE, max_tokens: int = DEFAULT_MAX_TOKENS, max_cost: TokenUsage | None = None, retry: RetryConfig | None = None, max_history_messages: int | None = None, trace: TraceCallback | None = None, approval_gate: ApprovalGate | None = None, redact_trace_args: bool = True, on_checkpoint: CheckpointCallback | None = None, summarizer: HistorySummarizer | None = None) -> PatternResult[str]

Execute a think-act-observe tool-calling loop.

The LLM is called repeatedly with the conversation history and available tool schemas. When the LLM returns tool calls, each tool is executed and its result appended as a tool-role message. The loop ends when the LLM responds without tool calls (final answer) or max_rounds is exhausted.

Supply either prompt (a single new user turn) or messages (a full prior conversation to continue), not both. Passing messages enables multi-turn assistants: the returned metadata["messages"] holds the complete updated transcript (the input history plus this run's assistant turns, tool results, and final answer) ready to feed into the next call.

Parameters:

Name	Type	Description	Default
provider	ToolCallingProvider	LLM provider to call.	required
prompt	str | None	Initial user prompt. Sugar for messages=[user_message(prompt)]; mutually exclusive with messages.	None
tools	Sequence[Tool]	Sequence of :class:Tool definitions available to the LLM.	()
messages	Sequence[dict[str, Any]] | None	A prior conversation (OpenAI-format message dicts) to continue instead of starting from prompt. The list is copied, never mutated.	None
max_rounds	int	Maximum think-act-observe cycles.	_DEFAULT_MAX_ROUNDS
max_observation_chars	int	Truncation limit for each tool result.	_DEFAULT_MAX_OBSERVATION_CHARS
tool_timeout	float | None	Per-call timeout override. Falls back to tool.timeout if None.	None
max_tool_calls_per_round	int	Ceiling on model-requested tool calls executed in a single round (they run concurrently, so this bounds the fan-out). Surplus calls are never executed; each receives a rejection observation telling the model to retry with fewer.	_DEFAULT_MAX_TOOL_CALLS_PER_ROUND
temperature	float	Sampling temperature (lower = more deterministic).	_DEFAULT_TEMPERATURE
max_tokens	int	Maximum tokens per LLM completion.	DEFAULT_MAX_TOKENS
max_cost	TokenUsage | None	Optional token/call budget.	None
retry	RetryConfig | None	Optional retry configuration per LLM call.	None
max_history_messages	int | None	When set, trim the message history to at most this many entries before each LLM call. Always keeps the first message (the original prompt). None disables trimming.	None
trace	TraceCallback | None	Optional structured trace callback.	None
approval_gate	ApprovalGate | None	Optional gate checked before each tool execution.	None
redact_trace_args	bool	When True (the default), argument values are replaced with "[redacted]" in tool_call_start trace events. Only argument keys are emitted, keeping traces useful for debugging without risking PII or credential leakage. Set to False to include raw argument values (e.g. in controlled test environments).	True
on_checkpoint	CheckpointCallback | None	Optional callback invoked after each round that dispatched tool calls, receiving (round, state) where round is 0-based and state is a JSON-serializable dict with keys round, last_response, tool_calls_made (list of tool names), cost, and messages (the full transcript so far, a list of message dicts). May be sync or async; exceptions are logged and swallowed so a failing checkpoint never aborts the loop.	None
summarizer	HistorySummarizer | None	Optional async callback used together with max_history_messages. When trimming drops earlier messages, the dropped messages (in order) are passed to summarizer and the returned text is injected as a system message into the per-round window sent to the provider, immediately after the preserved first message. The stored transcript is never modified — only the active window. None (the default) disables summarization, leaving trimming behaviour unchanged. The callback may return either the summary text or a (text, TokenUsage) pair; reported usage is folded into the loop's cost accounting and counted against max_cost on the following round. Summaries are memoized by the dropped-window boundary, so a window that is unchanged across rounds is summarized at most once rather than re-summarized each round.	None

Returns:

Name	Type	Description
A	PatternResult[str]	class:PatternResult whose value is the final LLM
	PatternResult[str]	response, score is None, and metadata includes
	PatternResult[str]	rounds and tool_calls_made.

Metadata

rounds (int): Number of think-act-observe cycles completed. tool_calls_made (int): Total individual tool invocations. rejected_tool_calls (int): Model-requested calls never executed because a round exceeded max_tool_calls_per_round. truncated_responses (int): LLM responses truncated due to finish_reason=length. truncated_observations (int): Tool results truncated due to max_observation_chars. messages_trimmed (int): Number of rounds where history was trimmed. summarized (int): Number of rounds where trimming dropped messages and a summarizer was supplied, so an earlier-conversation summary was injected into that round's active window. Always 0 when no summarizer is provided. messages (tuple[dict, ...]): The full conversation transcript after the loop, including the seeded input, every assistant/tool turn, and the final assistant answer. Feed back in via messages= to continue. termination_reason (TerminationReason | None): How the loop ended. TerminationReason.NATURAL when the LLM returned a final answer; TerminationReason.MAX_ITERATIONS when max_rounds was exhausted (also present on the raised :exc:~executionkit.provider.MaxIterationsError's .metadata). None only during error paths that abort early.

Source code in executionkit/patterns/react_loop.py

async def react_loop(
    provider: ToolCallingProvider,
    prompt: str | None = None,
    tools: Sequence[Tool] = (),
    *,
    messages: Sequence[dict[str, Any]] | None = None,
    max_rounds: int = _DEFAULT_MAX_ROUNDS,
    max_observation_chars: int = _DEFAULT_MAX_OBSERVATION_CHARS,
    tool_timeout: float | None = None,
    max_tool_calls_per_round: int = _DEFAULT_MAX_TOOL_CALLS_PER_ROUND,
    temperature: float = _DEFAULT_TEMPERATURE,
    max_tokens: int = DEFAULT_MAX_TOKENS,
    max_cost: TokenUsage | None = None,
    retry: RetryConfig | None = None,
    max_history_messages: int | None = None,
    trace: TraceCallback | None = None,
    approval_gate: ApprovalGate | None = None,
    redact_trace_args: bool = True,
    on_checkpoint: CheckpointCallback | None = None,
    summarizer: HistorySummarizer | None = None,
) -> PatternResult[str]:
    """Execute a think-act-observe tool-calling loop.

    The LLM is called repeatedly with the conversation history and
    available tool schemas.  When the LLM returns tool calls, each tool
    is executed and its result appended as a tool-role message.  The loop
    ends when the LLM responds without tool calls (final answer) or
    ``max_rounds`` is exhausted.

    Supply **either** *prompt* (a single new user turn) **or** *messages* (a
    full prior conversation to continue), not both.  Passing *messages* enables
    multi-turn assistants: the returned ``metadata["messages"]`` holds the
    complete updated transcript (the input history plus this run's assistant
    turns, tool results, and final answer) ready to feed into the next call.

    Args:
        provider: LLM provider to call.
        prompt: Initial user prompt. Sugar for ``messages=[user_message(prompt)]``;
            mutually exclusive with *messages*.
        tools: Sequence of :class:`Tool` definitions available to the LLM.
        messages: A prior conversation (OpenAI-format message dicts) to continue
            instead of starting from *prompt*. The list is copied, never mutated.
        max_rounds: Maximum think-act-observe cycles.
        max_observation_chars: Truncation limit for each tool result.
        tool_timeout: Per-call timeout override.  Falls back to
            ``tool.timeout`` if ``None``.
        max_tool_calls_per_round: Ceiling on model-requested tool calls
            executed in a single round (they run concurrently, so this bounds
            the fan-out). Surplus calls are never executed; each receives a
            rejection observation telling the model to retry with fewer.
        temperature: Sampling temperature (lower = more deterministic).
        max_tokens: Maximum tokens per LLM completion.
        max_cost: Optional token/call budget.
        retry: Optional retry configuration per LLM call.
        max_history_messages: When set, trim the message history to at most
            this many entries before each LLM call. Always keeps the first
            message (the original prompt). ``None`` disables trimming.
        trace: Optional structured trace callback.
        approval_gate: Optional gate checked before each tool execution.
        redact_trace_args: When ``True`` (the default), argument *values* are
            replaced with ``"[redacted]"`` in ``tool_call_start`` trace events.
            Only argument *keys* are emitted, keeping traces useful for
            debugging without risking PII or credential leakage.  Set to
            ``False`` to include raw argument values (e.g. in controlled test
            environments).
        on_checkpoint: Optional callback invoked after each round that dispatched
            tool calls, receiving ``(round, state)`` where ``round`` is 0-based
            and ``state`` is a JSON-serializable dict with keys ``round``,
            ``last_response``, ``tool_calls_made`` (list of tool names),
            ``cost``, and ``messages`` (the full transcript so far, a list of
            message dicts). May be sync or async; exceptions are logged and
            swallowed so a failing checkpoint never aborts the loop.
        summarizer: Optional async callback used together with
            ``max_history_messages``. When trimming drops earlier messages, the
            dropped messages (in order) are passed to ``summarizer`` and the
            returned text is injected as a system message into the *per-round*
            window sent to the provider, immediately after the preserved first
            message. The stored transcript is never modified — only the active
            window. ``None`` (the default) disables summarization, leaving
            trimming behaviour unchanged.

            The callback may return either the summary text or a
            ``(text, TokenUsage)`` pair; reported usage is folded into the
            loop's cost accounting and counted against ``max_cost`` on the
            following round. Summaries are memoized by the dropped-window
            boundary, so a window that is unchanged across rounds is summarized
            at most once rather than re-summarized each round.

    Returns:
        A :class:`PatternResult` whose ``value`` is the final LLM
        response, ``score`` is ``None``, and ``metadata`` includes
        ``rounds`` and ``tool_calls_made``.

    Metadata:
        rounds (int): Number of think-act-observe cycles completed.
        tool_calls_made (int): Total individual tool invocations.
        rejected_tool_calls (int): Model-requested calls never executed because
            a round exceeded ``max_tool_calls_per_round``.
        truncated_responses (int): LLM responses truncated due to
            ``finish_reason=length``.
        truncated_observations (int): Tool results truncated due to
            ``max_observation_chars``.
        messages_trimmed (int): Number of rounds where history was trimmed.
        summarized (int): Number of rounds where trimming dropped messages and a
            ``summarizer`` was supplied, so an earlier-conversation summary was
            injected into that round's active window. Always 0 when no
            ``summarizer`` is provided.
        messages (tuple[dict, ...]): The full conversation transcript after the
            loop, including the seeded input, every assistant/tool turn, and the
            final assistant answer. Feed back in via ``messages=`` to continue.
        termination_reason (TerminationReason | None): How the loop ended.
            ``TerminationReason.NATURAL`` when the LLM returned a final
            answer; ``TerminationReason.MAX_ITERATIONS`` when ``max_rounds``
            was exhausted (also present on the raised
            :exc:`~executionkit.provider.MaxIterationsError`'s ``.metadata``).
            ``None`` only during error paths that abort early.
    """
    _validate_react_loop_args(
        provider,
        max_rounds,
        max_observation_chars,
        tool_timeout,
        max_tokens,
        max_history_messages,
        max_tool_calls_per_round,
    )
    tracker = CostTracker()
    metadata: dict[str, Any] = {
        "rounds": 0,
        "tool_calls_made": 0,
        "rejected_tool_calls": 0,
        "truncated_responses": 0,
        "truncated_observations": 0,
        "messages_trimmed": 0,
        "summarized": 0,
        "termination_reason": None,
    }
    tool_schemas = [tool.to_schema() for tool in tools]
    tool_lookup: dict[str, Tool] = {tool.name: tool for tool in tools}
    history = _seed_messages(prompt, messages)
    # Memoizes summaries by dropped-window boundary so a stable window is
    # summarized at most once even when trimming recurs across rounds.
    summary_cache: dict[int, str] = {}

    for round_num in range(1, max_rounds + 1):
        if max_history_messages is not None:
            active_messages = _trim_messages(history, max_history_messages)
            if len(active_messages) < len(history):
                metadata["messages_trimmed"] = int(metadata["messages_trimmed"]) + 1
                if summarizer is not None:
                    active_messages = await _summarize_trimmed_window(
                        history,
                        active_messages,
                        summarizer,
                        tracker,
                        summary_cache,
                    )
                    metadata["summarized"] = int(metadata["summarized"]) + 1
        else:
            active_messages = history
        response = await checked_complete(
            provider,
            active_messages,
            tracker,
            max_cost,
            retry,
            trace,
            temperature=temperature,
            max_tokens=max_tokens,
            tools=tool_schemas,
        )
        _note_truncation(response, metadata, "react_loop")
        metadata["rounds"] = round_num

        # No tool calls means the LLM is done — return the content.
        if not response.has_tool_calls:
            metadata["termination_reason"] = TerminationReason.NATURAL
            history.append(assistant_message(response.content))
            return PatternResult[str](
                value=response.content,
                score=None,
                cost=tracker.to_usage(),
                metadata=MappingProxyType({**metadata, "messages": tuple(history)}),
            )

        # Append assistant message with tool calls to conversation
        history.append(_build_assistant_message(response))

        # Execute each tool call and append results
        await _execute_tool_calls_round(
            response.tool_calls,
            tool_lookup,
            tool_timeout,
            max_observation_chars,
            metadata,
            history,
            max_tool_calls_per_round=max_tool_calls_per_round,
            trace=trace,
            approval_gate=approval_gate,
            redact_trace_args=redact_trace_args,
        )

        if on_checkpoint is not None:
            checkpoint_state: dict[str, Any] = {
                "round": round_num - 1,
                "last_response": response.content,
                "tool_calls_made": [tc.name for tc in response.tool_calls],
                "cost": dataclasses.asdict(tracker.snapshot()),
                "messages": list(history),
            }
            await run_checkpoint(
                on_checkpoint, round_num - 1, checkpoint_state, context="react_loop"
            )

    metadata["termination_reason"] = TerminationReason.MAX_ITERATIONS
    metadata["messages"] = tuple(history)
    raise MaxIterationsError(
        "react_loop() reached max_rounds without a final answer",
        cost=tracker.to_usage(),
        metadata=dict(metadata),
    )

structured

executionkit.patterns.structured.structured async

structured(provider: LLMProvider, prompt: str, *, validator: StructuredValidator | None = None, max_retries: int = 3, temperature: float = 0.0, max_tokens: int = 4096, max_cost: TokenUsage | None = None, retry: RetryConfig | None = None, trace: TraceCallback | None = None, stream: bool = False) -> PatternResult[StructuredValue]

Request JSON output, parse it, and optionally repair invalid responses.

Validators should return None, True, or "" for success. Any other value is treated as a validation failure and included in the repair prompt. max_retries=0 is supported and means "make one parse attempt with no repair call".

Source code in executionkit/patterns/structured.py

async def structured(
    provider: LLMProvider,
    prompt: str,
    *,
    validator: StructuredValidator | None = None,
    max_retries: int = 3,
    temperature: float = 0.0,
    max_tokens: int = 4096,
    max_cost: TokenUsage | None = None,
    retry: RetryConfig | None = None,
    trace: TraceCallback | None = None,
    stream: bool = False,
) -> PatternResult[StructuredValue]:
    """Request JSON output, parse it, and optionally repair invalid responses.

    Validators should return ``None``, ``True``, or ``""`` for success. Any
    other value is treated as a validation failure and included in the repair
    prompt. ``max_retries=0`` is supported and means "make one parse attempt
    with no repair call".
    """
    if stream:
        raise ValueError(
            "stream=True is not supported for structured: this pattern "
            "aggregates intermediate results before returning."
        )
    if max_retries < 0:
        raise ValueError(f"max_retries must be >= 0, got {max_retries}")
    if max_tokens < 1:
        raise ValueError(f"max_tokens must be >= 1, got {max_tokens}")

    tracker = CostTracker()
    metadata: dict[str, Any] = {
        "parse_attempts": 0,
        "repair_attempts": 0,
        "validated": validator is None,
    }

    response = await checked_complete(
        provider,
        [user_message(_json_prompt(prompt))],
        tracker,
        max_cost,
        retry,
        trace,
        temperature=temperature,
        max_tokens=max_tokens,
    )
    latest_text = response.content
    last_error = "Structured output could not be produced."

    for attempt in range(max_retries + 1):
        metadata["parse_attempts"] = attempt + 1
        try:
            value = extract_json(latest_text)
        except ValueError as exc:
            last_error = f"JSON parse failed: {exc}"
        else:
            validation_error = (
                None
                if validator is None
                else _normalize_validation_error(validator(value))
            )
            if validation_error is None:
                metadata["validated"] = True
                return PatternResult(
                    value=value,
                    cost=tracker.to_usage(),
                    metadata=MappingProxyType(dict(metadata)),
                )
            last_error = validation_error

        if attempt == max_retries:
            break

        metadata["repair_attempts"] += 1
        repair_prompt = (
            "The previous response was not valid structured output.\n"
            f"Error: {last_error}\n\n"
            "Original task:\n"
            f"{prompt}\n\n"
            "Previous response:\n"
            f"{latest_text}\n\n"
            "Return a corrected JSON object or array only."
        )
        repair_response = await checked_complete(
            provider,
            [user_message(repair_prompt)],
            tracker,
            max_cost,
            retry,
            trace,
            temperature=temperature,
            max_tokens=max_tokens,
        )
        latest_text = repair_response.content

    raise PatternError(last_error, cost=tracker.to_usage())

pipe

executionkit.compose.pipe async

pipe(provider: LLMProvider, prompt: str, *steps: PatternStep, max_budget: TokenUsage | None = None, **shared_kwargs: Any) -> PatternResult[Any]

Chain reasoning patterns, threading output as the next prompt.

Each step must be an async callable with the signature::

async def step(provider, prompt, **kwargs) -> PatternResult[Any]

The value of each result is converted to a string and passed as the prompt to the following step. Costs are accumulated and, when max_budget is given, the remaining budget is forwarded to each step via the max_cost keyword argument.

Parameters:

Name	Type	Description	Default
provider	LLMProvider	LLM provider passed unchanged to every step.	required
prompt	str	Initial input prompt.	required
*steps	PatternStep	Async pattern callables to chain in order.	()
max_budget	TokenUsage | None	Optional shared token/call budget across all steps.	None
**shared_kwargs	Any	Extra keyword arguments forwarded to every step.	{}

Returns:

Name	Type	Description
The	PatternResult[Any]	class:~executionkit.types.PatternResult from the final step,
	PatternResult[Any]	with its cost replaced by the cumulative cost across all steps.
	PatternResult[Any]	If steps is empty the prompt is returned as-is with zero cost.
	PatternResult[Any]	The result metadata gains step_costs: a tuple of per-step
	PatternResult[Any]	class:~executionkit.types.TokenUsage deltas in execution order. On the
	PatternResult[Any]	error path the same step_costs (including the failing step's partial
	PatternResult[Any]	spend) is attached to the re-raised exception's metadata.

Source code in executionkit/compose.py

async def pipe(
    provider: LLMProvider,
    prompt: str,
    *steps: PatternStep,
    max_budget: TokenUsage | None = None,
    **shared_kwargs: Any,
) -> PatternResult[Any]:
    """Chain reasoning patterns, threading output as the next prompt.

    Each *step* must be an async callable with the signature::

        async def step(provider, prompt, **kwargs) -> PatternResult[Any]

    The ``value`` of each result is converted to a string and passed as the
    *prompt* to the following step.  Costs are accumulated and, when
    *max_budget* is given, the remaining budget is forwarded to each step
    via the ``max_cost`` keyword argument.

    Args:
        provider: LLM provider passed unchanged to every step.
        prompt: Initial input prompt.
        *steps: Async pattern callables to chain in order.
        max_budget: Optional shared token/call budget across all steps.
        **shared_kwargs: Extra keyword arguments forwarded to every step.

    Returns:
        The :class:`~executionkit.types.PatternResult` from the final step,
        with its ``cost`` replaced by the cumulative cost across all steps.
        If *steps* is empty the prompt is returned as-is with zero cost.

        The result ``metadata`` gains ``step_costs``: a ``tuple`` of per-step
        :class:`~executionkit.types.TokenUsage` deltas in execution order. On the
        error path the same ``step_costs`` (including the failing step's partial
        spend) is attached to the re-raised exception's ``metadata``.
    """
    if not steps:
        return PatternResult(value=prompt)

    total_cost = TokenUsage()
    current_prompt: str = prompt
    last_result: PatternResult[Any] | None = None
    step_metadata: list[dict[str, Any]] = []
    step_costs: list[TokenUsage] = []

    for step in steps:
        step_kwargs = dict(shared_kwargs)
        if max_budget is not None:
            step_kwargs["max_cost"] = _subtract(max_budget, total_cost)
        filtered_kwargs = _filter_kwargs(step, step_kwargs)

        try:
            result: PatternResult[Any] = await step(
                provider, current_prompt, **filtered_kwargs
            )
        except ExecutionKitError as exc:
            # Record the failing step's partial spend, then raise a shallow copy
            # carrying the accumulated cost.  The original exception is NOT
            # mutated (immutability): a fresh metadata dict is assigned, and the
            # cause chain and traceback are preserved.
            step_costs.append(exc.cost)
            new_exc = copy.copy(exc)
            new_exc.cost = total_cost + exc.cost
            new_exc.metadata = {**exc.metadata, "step_costs": tuple(step_costs)}
            raise new_exc.with_traceback(exc.__traceback__) from exc.__cause__

        step_costs.append(result.cost)
        total_cost = total_cost + result.cost
        current_prompt = str(result.value)
        step_metadata.append(dict(result.metadata))
        last_result = result

    assert last_result is not None  # noqa: S101  # guarded by early return above
    final_metadata: dict[str, Any] = dict(last_result.metadata)
    final_metadata["step_count"] = len(steps)
    final_metadata["step_metadata"] = step_metadata
    final_metadata["step_costs"] = tuple(step_costs)
    return PatternResult(
        value=last_result.value,
        score=last_result.score,
        cost=total_cost,
        metadata=MappingProxyType(final_metadata),
    )

executionkit.compose.PatternStep

Bases: Protocol

Callable protocol for a single step in a :func:pipe chain.

Each step must accept (provider, prompt, **kwargs) and return an awaitable :class:~executionkit.types.PatternResult. Extra keyword arguments (e.g. max_cost) are filtered to only those the step actually accepts, so steps that do not declare **kwargs will not receive unsupported arguments.

Sync wrappers

from executionkit import (
    consensus_sync,
    refine_loop_sync,
    react_loop_sync,
    structured_sync,
    pipe_sync,
)

Each sync wrapper takes the same arguments as its async counterpart and runs it via asyncio.run. They raise RuntimeError when called inside a running event loop — use await directly there.

Value types

executionkit.types.PatternResult dataclass

PatternResult(value: T, score: float | None = None, cost: TokenUsage = TokenUsage(), metadata: MappingProxyType[str, Any] = (lambda: MappingProxyType({}))())

Bases: Generic[T]

Result returned by every reasoning pattern.

metadata keys vary by pattern. Each pattern documents its own keys in its function docstring. Do not rely on undocumented keys — they are private.

executionkit.types.TokenUsage dataclass

TokenUsage(input_tokens: int = 0, output_tokens: int = 0, llm_calls: int = 0)

Accumulated token and call counts.

__sub__

__sub__(other: TokenUsage) -> TokenUsage

Return the field-wise difference self - other.

Useful for computing the delta between two :class:CostTracker snapshots (e.g. per-step cost attribution). No clamping is applied, so callers that subtract a later snapshot from an earlier one can observe negative fields.

Source code in executionkit/types.py

def __sub__(self, other: TokenUsage) -> TokenUsage:
    """Return the field-wise difference ``self - other``.

    Useful for computing the delta between two :class:`CostTracker`
    snapshots (e.g. per-step cost attribution).  No clamping is applied,
    so callers that subtract a later snapshot from an earlier one can
    observe negative fields.
    """
    return TokenUsage(
        input_tokens=self.input_tokens - other.input_tokens,
        output_tokens=self.output_tokens - other.output_tokens,
        llm_calls=self.llm_calls - other.llm_calls,
    )

executionkit.types.Tool dataclass

Tool(name: str, description: str, parameters: Mapping[str, Any], execute: Callable[..., Awaitable[str]], timeout: float = DEFAULT_TOOL_TIMEOUT_SECONDS)

Describes a tool available for LLM tool-calling.

parameters is a JSON Schema mapping describing the function arguments. Automatically wrapped in a read-only proxy. execute is the async callable invoked when the LLM requests this tool.

__post_init__

__post_init__() -> None

Wrap parameters in a read-only proxy to enforce immutability.

Source code in executionkit/types.py

def __post_init__(self) -> None:
    """Wrap ``parameters`` in a read-only proxy to enforce immutability."""
    if not isinstance(self.parameters, MappingProxyType):
        object.__setattr__(
            self, "parameters", MappingProxyType(dict(self.parameters))
        )

to_schema

to_schema() -> dict[str, Any]

Return the OpenAI-compatible function tool schema.

Source code in executionkit/types.py

def to_schema(self) -> dict[str, Any]:
    """Return the OpenAI-compatible function tool schema."""
    return {
        "type": "function",
        "function": {
            "name": self.name,
            "description": self.description,
            "parameters": dict(self.parameters),
        },
    }

executionkit.types.VotingStrategy

Bases: StrEnum

Strategy for consensus voting.

executionkit.types.Evaluator module-attribute

Evaluator: TypeAlias = Callable[[str, 'LLMProvider'], Awaitable[float]]

Async callable that scores a response string on [0.0, 1.0].

Evals

executionkit.evals.EvalCase dataclass

EvalCase(name: str, run: EvalRun, check: EvalCheck, metadata: Mapping[str, Any] = dict())

A single eval case with a runner and a result check.

executionkit.evals.EvalResult dataclass

EvalResult(name: str, passed: bool, reason: str = '', metadata: MappingProxyType[str, Any] = (lambda: MappingProxyType({}))())

Outcome of a single eval case.

executionkit.evals.EvalReport dataclass

EvalReport(results: tuple[EvalResult, ...], min_accuracy: float | None = None)

Aggregate eval report.

Attributes:

Name	Type	Description
results	tuple[EvalResult, ...]	Per-case outcomes.
min_accuracy	float | None	Optional accuracy threshold used by :func:run_eval_suite for live / non-deterministic suites. When set, callers gate on :attr:accuracy_passed instead of :attr:passed to allow a small number of tolerated failures.

accuracy property

accuracy: float

Fraction of cases that passed; 0.0 when the suite is empty.

accuracy_passed property

accuracy_passed: bool

True when accuracy >= min_accuracy (live-suite gate).

Falls back to :attr:passed when no min_accuracy was configured, so callers can always use this property as the single gate regardless of suite type.

passed property

passed: bool

True when every case passed (100% — required for deterministic suites).

summary

summary() -> str

One-line human-readable result, e.g. '7/9 passed (77.8% accuracy)'.

Source code in executionkit/evals.py

def summary(self) -> str:
    """One-line human-readable result, e.g. '7/9 passed (77.8% accuracy)'."""
    return f"{self.passed_count}/{self.total} passed ({self.accuracy:.1%} accuracy)"

executionkit.evals.run_eval_suite async

run_eval_suite(cases: Sequence[EvalCase], *, min_accuracy: float | None = None) -> EvalReport

Run eval cases in order and return pass/fail results.

Parameters:

Name	Type	Description	Default
cases	Sequence[EvalCase]	Eval cases to run.	required
min_accuracy	float | None	When provided, the report is considered passing when report.accuracy >= min_accuracy rather than requiring every case to pass (report.passed). Use :data:LIVE_EVAL_MIN_ACCURACY for non-deterministic / live-provider suites that tolerate a small number of failures. Deterministic golden suites should omit this parameter so they continue to require 100% pass rate via report.passed.	None

Returns:

Type	Description
EvalReport	class:EvalReport with per-case results. The caller gates on either
EvalReport	report.passed (deterministic) or
EvalReport	report.accuracy >= min_accuracy (live).

Source code in executionkit/evals.py

async def run_eval_suite(
    cases: Sequence[EvalCase],
    *,
    min_accuracy: float | None = None,
) -> EvalReport:
    """Run eval cases in order and return pass/fail results.

    Args:
        cases: Eval cases to run.
        min_accuracy: When provided, the report is considered passing when
            ``report.accuracy >= min_accuracy`` rather than requiring every
            case to pass (``report.passed``).  Use :data:`LIVE_EVAL_MIN_ACCURACY`
            for non-deterministic / live-provider suites that tolerate a small
            number of failures.  Deterministic golden suites should omit this
            parameter so they continue to require 100% pass rate via
            ``report.passed``.

    Returns:
        :class:`EvalReport` with per-case results.  The caller gates on either
        ``report.passed`` (deterministic) or
        ``report.accuracy >= min_accuracy`` (live).
    """
    return EvalReport(
        results=tuple([await _run_case(case) for case in cases]),
        min_accuracy=min_accuracy,
    )

executionkit.evals.live_provider_from_env

live_provider_from_env() -> Provider | None

Build an opt-in live eval provider from EXECUTIONKIT_* env vars.

Returns None unless EXECUTIONKIT_LIVE_EVAL=1. When enabled, EXECUTIONKIT_BASE_URL and EXECUTIONKIT_MODEL are required; EXECUTIONKIT_API_KEY is optional and defaults to an empty string.

Source code in executionkit/evals.py

def live_provider_from_env() -> Provider | None:
    """Build an opt-in live eval provider from EXECUTIONKIT_* env vars.

    Returns ``None`` unless ``EXECUTIONKIT_LIVE_EVAL=1``. When enabled,
    ``EXECUTIONKIT_BASE_URL`` and ``EXECUTIONKIT_MODEL`` are required;
    ``EXECUTIONKIT_API_KEY`` is optional and defaults to an empty string.
    """

    if os.getenv("EXECUTIONKIT_LIVE_EVAL") != "1":
        return None

    base_url = os.getenv("EXECUTIONKIT_BASE_URL")
    model = os.getenv("EXECUTIONKIT_MODEL")
    missing = [
        name
        for name, value in (
            ("EXECUTIONKIT_BASE_URL", base_url),
            ("EXECUTIONKIT_MODEL", model),
        )
        if not value
    ]
    if missing:
        raise ValueError(f"{', '.join(missing)} required for live evals")

    return Provider(
        base_url=base_url or "",
        model=model or "",
        api_key=os.getenv("EXECUTIONKIT_API_KEY", ""),
    )

Observability

executionkit.observability.TraceEvent dataclass

TraceEvent(kind: str, payload: MappingProxyType[str, Any] = (lambda: MappingProxyType({}))())

A structured event emitted by patterns and lightweight primitives.

executionkit.observability.TraceCallback module-attribute

TraceCallback: TypeAlias = Callable[['TraceEvent'], Awaitable[None] | None]

executionkit.observability.emit_trace async

emit_trace(trace: TraceCallback | None, event: TraceEvent) -> None

Emit event to an optional callback.

Source code in executionkit/observability.py

async def emit_trace(trace: TraceCallback | None, event: TraceEvent) -> None:
    """Emit *event* to an optional callback."""

    if trace is None:
        return
    maybe_awaitable = trace(event)
    if inspect.isawaitable(maybe_awaitable):
        await maybe_awaitable

Routing

executionkit.routing.RouteRule dataclass

RouteRule(name: str, provider: LLMProvider, predicate: RoutePredicate, metadata: Mapping[str, Any] = dict())

A named provider selection rule.

executionkit.routing.Router

Router(*, rules: Sequence[RouteRule], fallback: LLMProvider)

Select a provider by evaluating rules before a pattern call.

Source code in executionkit/routing.py

def __init__(self, *, rules: Sequence[RouteRule], fallback: LLMProvider) -> None:
    self.rules = tuple(rules)
    self.fallback = fallback

run async

run(pattern: RoutedPattern, prompt: str, *, context: Mapping[str, Any] | None = None, **kwargs: Any) -> PatternResult[Any]

Select a provider from context, then call pattern with it.

Routing inputs are passed explicitly via context and forwarded only to the route predicates; **kwargs are forwarded only to pattern. Keeping the two disjoint stops routing keys (e.g. tier) from leaking into the pattern call — which would raise TypeError for any pattern that does not declare them.

Source code in executionkit/routing.py

async def run(
    self,
    pattern: RoutedPattern,
    prompt: str,
    *,
    context: Mapping[str, Any] | None = None,
    **kwargs: Any,
) -> PatternResult[Any]:
    """Select a provider from *context*, then call *pattern* with it.

    Routing inputs are passed explicitly via *context* and forwarded only to
    the route predicates; ``**kwargs`` are forwarded only to *pattern*.
    Keeping the two disjoint stops routing keys (e.g. ``tier``) from leaking
    into the pattern call — which would raise ``TypeError`` for any pattern
    that does not declare them.
    """
    # Drop a stray "prompt" key so it cannot collide with the positional
    # ``prompt`` argument to ``select`` (``select(prompt, prompt=...)`` would
    # raise TypeError). The predicate still receives the real prompt
    # positionally, so nothing is lost.
    route_context = {
        key: value for key, value in (context or {}).items() if key != "prompt"
    }
    provider = self.select(prompt, **route_context)
    return await pattern(provider, prompt, **kwargs)

Workflow and planning

executionkit.workflow.Step dataclass

Step(name: str, run: WorkflowRun, depends_on: tuple[str, ...] = (), metadata: Mapping[str, Any] = dict())

A named workflow step with optional dependencies.

executionkit.workflow.Workflow

Workflow(steps: Sequence[Step])

Run named steps once their dependencies are available.

Source code in executionkit/workflow.py

def __init__(self, steps: Sequence[Step]) -> None:
    self.steps = tuple(steps)
    self._validate()

run async

run(initial_context: Mapping[str, Any] | None = None, *, trace: TraceCallback | None = None, approval_gate: ApprovalGate | None = None, checkpoint_fn: CheckpointFn | None = None, resume_from: WorkflowCheckpoint | None = None) -> WorkflowResult

Execute the workflow, optionally checkpointing after each batch.

Parameters

initial_context: Key/value pairs injected into the step context before execution. trace: Optional async or sync callback receiving :class:~executionkit.observability.TraceEvent objects. approval_gate: Optional gate consulted before each step runs. checkpoint_fn: Called with a :class:WorkflowCheckpoint after each batch of completed steps. The caller is responsible for persisting the checkpoint; this library imposes no storage requirement. resume_from: A previously saved :class:WorkflowCheckpoint. Steps whose names already appear in resume_from.outputs are skipped; accumulated outputs and token budget are restored verbatim. When None (default), the workflow starts from the beginning.

Source code in executionkit/workflow.py

async def run(
    self,
    initial_context: Mapping[str, Any] | None = None,
    *,
    trace: TraceCallback | None = None,
    approval_gate: ApprovalGate | None = None,
    checkpoint_fn: CheckpointFn | None = None,
    resume_from: WorkflowCheckpoint | None = None,
) -> WorkflowResult:
    """Execute the workflow, optionally checkpointing after each batch.

    Parameters
    ----------
    initial_context:
        Key/value pairs injected into the step context before execution.
    trace:
        Optional async or sync callback receiving
        :class:`~executionkit.observability.TraceEvent` objects.
    approval_gate:
        Optional gate consulted before each step runs.
    checkpoint_fn:
        Called with a :class:`WorkflowCheckpoint` after each batch of
        completed steps.  The caller is responsible for persisting the
        checkpoint; this library imposes no storage requirement.
    resume_from:
        A previously saved :class:`WorkflowCheckpoint`.  Steps whose
        names already appear in ``resume_from.outputs`` are skipped;
        accumulated outputs and token budget are restored verbatim.
        When ``None`` (default), the workflow starts from the beginning.
    """
    # ------------------------------------------------------------------
    # Restore state from checkpoint (if any)
    # ------------------------------------------------------------------
    if resume_from is not None:
        outputs: dict[str, Any] = dict(resume_from.outputs)
        total_cost = resume_from.cost
    else:
        outputs = dict(initial_context or {})
        total_cost = TokenUsage()

    # Steps whose name already exists in outputs are already done.
    pending = {step.name: step for step in self.steps if step.name not in outputs}
    completed_count = len(self.steps) - len(pending)

    while pending:
        ready = [
            step
            for step in pending.values()
            if all(dep in outputs for dep in step.depends_on)
        ]
        if not ready:
            raise ExecutionKitError("Workflow dependencies could not be resolved")

        results = await gather_strict(
            [
                self._run_step(
                    step,
                    outputs,
                    trace=trace,
                    approval_gate=approval_gate,
                )
                for step in ready
            ]
        )
        for step, output in zip(ready, results, strict=True):
            if isinstance(output, PatternResult):
                total_cost += output.cost
                outputs[step.name] = output.value
            else:
                outputs[step.name] = output
            pending.pop(step.name)

        completed_count += len(ready)

        if checkpoint_fn is not None:
            maybe_checkpoint = checkpoint_fn(
                WorkflowCheckpoint(
                    step_index=completed_count,
                    outputs=MappingProxyType(dict(outputs)),
                    cost=total_cost,
                )
            )
            if inspect.isawaitable(maybe_checkpoint):
                await maybe_checkpoint

    return WorkflowResult(outputs=MappingProxyType(outputs), cost=total_cost)

executionkit.workflow.WorkflowResult dataclass

WorkflowResult(outputs: MappingProxyType[str, Any], cost: TokenUsage = TokenUsage())

Outputs and aggregate cost from a workflow run.

executionkit.planning.PlanStep dataclass

PlanStep(name: str, instruction: str, run: PlanRun, metadata: Mapping[str, Any] = dict())

A human-readable executable plan step.

executionkit.planning.Plan

Plan(steps: Sequence[PlanStep])

Execute named plan steps in order.

Source code in executionkit/planning.py

def __init__(self, steps: Sequence[PlanStep]) -> None:
    names = [step.name for step in steps]
    if len(names) != len(set(names)):
        raise ValueError("Plan step names must be unique")
    self.steps = tuple(steps)

executionkit.planning.PlanResult dataclass

PlanResult(outputs: MappingProxyType[str, Any], cost: TokenUsage = TokenUsage())

Outputs and aggregate cost from plan execution.

Approval

executionkit.approval.ApprovalRequest dataclass

ApprovalRequest(action: str, subject: str, metadata: MappingProxyType[str, Any] = (lambda: MappingProxyType({}))())

A request to approve an operation before it runs.

executionkit.approval.ApprovalDecision dataclass

ApprovalDecision(approved: bool, reason: str = '')

Approval result.

executionkit.approval.ApprovalGate

ApprovalGate(callback: ApprovalCallback, *, timeout_seconds: float | None = None, on_timeout: Literal['approve', 'deny', 'raise'] = 'raise')

Async-compatible approval callback wrapper.

Parameters

callback: Synchronous or async callable that receives an :class:ApprovalRequest and returns an :class:ApprovalDecision (or a truthy/falsy value). timeout_seconds: If given, the callback must resolve within this many seconds. When it does not, on_timeout controls the fallback behaviour. None (default) preserves the original blocking behaviour. on_timeout: What to do when timeout_seconds elapses before the callback resolves.

* ``"raise"`` (default) — raise :class:`ApprovalTimeoutError`.
* ``"approve"`` — treat the timed-out request as approved.
* ``"deny"``   — treat the timed-out request as denied.

.. warning:: Setting on_timeout="approve" creates a fail-open gate: if the approval callback does not respond in time, the operation is automatically permitted. This is a potential privilege-escalation vector. Only use this option when you have deliberately decided that availability is more important than security for the guarded operation.

Source code in executionkit/approval.py

def __init__(
    self,
    callback: ApprovalCallback,
    *,
    timeout_seconds: float | None = None,
    on_timeout: Literal["approve", "deny", "raise"] = "raise",
) -> None:
    if on_timeout == "approve":
        warnings.warn(
            self._APPROVE_ON_TIMEOUT_WARNING,
            UserWarning,
            stacklevel=2,
        )
    self._callback = callback
    self._timeout_seconds = timeout_seconds
    self._on_timeout = on_timeout

executionkit.approval.ApprovalDeniedError

ApprovalDeniedError(message: str, *, cost: TokenUsage | None = None, metadata: dict[str, Any] | None = None)

Bases: ExecutionKitError

Raised when an approval-gated operation is denied.

Source code in executionkit/errors.py

def __init__(
    self,
    message: str,
    *,
    cost: TokenUsage | None = None,
    metadata: dict[str, Any] | None = None,
) -> None:
    super().__init__(message)
    self.cost: TokenUsage = cost if cost is not None else TokenUsage()
    self.metadata: dict[str, Any] = metadata if metadata is not None else {}

Session

executionkit.kit.Kit

Kit(provider: LLMProvider, *, track_cost: bool = True, messages: Sequence[dict[str, Any]] | None = None, rate_limiter: TokenBucket | None = None)

Session that holds a :class:~executionkit.provider.Provider and tracks cumulative token usage across all pattern calls.

Parameters:

Name	Type	Description	Default
provider	LLMProvider	The LLM provider to use for all calls.	required
track_cost	bool	When True (default), accumulate usage in an internal :class:~executionkit.cost.CostTracker. Set to False to disable tracking (e.g. in hot paths or tests).	True
messages	Sequence[dict[str, Any]] | None	Optional seed conversation (OpenAI-format message dicts) for multi-turn use via :meth:turn. Copied on construction.	None
rate_limiter	TokenBucket | None	Optional :class:~executionkit.engine.rate_bucket.TokenBucket. When provided, every pattern dispatch first awaits one token, pacing calls to a sustained rate (and honouring any Retry-After penalty). Defaults to None (unlimited).	None

Conversation state

:attr:messages holds the running transcript across :meth:turn calls. The single-shot pattern methods (:meth:react, :meth:consensus, …) do not read or write it — use :meth:turn for stateful conversations.

Source code in executionkit/kit.py

def __init__(
    self,
    provider: LLMProvider,
    *,
    track_cost: bool = True,
    messages: Sequence[dict[str, Any]] | None = None,
    rate_limiter: TokenBucket | None = None,
) -> None:
    self.provider = provider
    self._tracker: CostTracker | None = CostTracker() if track_cost else None
    self.messages: list[dict[str, Any]] = list(messages) if messages else []
    self._rate_limiter = rate_limiter

usage property

usage: TokenUsage

Cumulative token usage across all calls made through this Kit.

consensus async

consensus(prompt: str, **kwargs: Any) -> PatternResult[str]

Run the :func:~executionkit.patterns.consensus.consensus pattern.

All keyword arguments are forwarded unchanged to :func:consensus.

Source code in executionkit/kit.py

async def consensus(self, prompt: str, **kwargs: Any) -> PatternResult[str]:
    """Run the :func:`~executionkit.patterns.consensus.consensus` pattern.

    All keyword arguments are forwarded unchanged to :func:`consensus`.
    """
    return await self._run_tracked(consensus(self.provider, prompt, **kwargs))

map_reduce async

map_reduce(inputs: Sequence[str], **kwargs: Any) -> PatternResult[str]

Run the :func:~executionkit.patterns.map_reduce.map_reduce pattern.

All keyword arguments are forwarded unchanged to :func:map_reduce. map_prompt_template and reduce_prompt_template are required keyword arguments.

Source code in executionkit/kit.py

async def map_reduce(
    self, inputs: Sequence[str], **kwargs: Any
) -> PatternResult[str]:
    """Run the :func:`~executionkit.patterns.map_reduce.map_reduce` pattern.

    All keyword arguments are forwarded unchanged to :func:`map_reduce`.
    ``map_prompt_template`` and ``reduce_prompt_template`` are required
    keyword arguments.
    """
    return await self._run_tracked(map_reduce(self.provider, inputs, **kwargs))

pipe async

pipe(prompt: str, *steps: Callable[..., Any], **kwargs: Any) -> PatternResult[Any]

Run :func:~executionkit.compose.pipe with this Kit's provider.

All keyword arguments are forwarded unchanged to :func:pipe.

Source code in executionkit/kit.py

async def pipe(
    self, prompt: str, *steps: Callable[..., Any], **kwargs: Any
) -> PatternResult[Any]:
    """Run :func:`~executionkit.compose.pipe` with this Kit's provider.

    All keyword arguments are forwarded unchanged to :func:`pipe`.
    """
    return await self._run_tracked(pipe(self.provider, prompt, *steps, **kwargs))

react async

react(prompt: str, tools: Sequence[Tool], **kwargs: Any) -> PatternResult[str]

Run the :func:~executionkit.patterns.react_loop.react_loop pattern.

All keyword arguments are forwarded unchanged to :func:react_loop. The provider must satisfy :class:~executionkit.provider.ToolCallingProvider; a :exc:TypeError is raised if it does not.

Source code in executionkit/kit.py

async def react(
    self, prompt: str, tools: Sequence[Tool], **kwargs: Any
) -> PatternResult[str]:
    """Run the :func:`~executionkit.patterns.react_loop.react_loop` pattern.

    All keyword arguments are forwarded unchanged to :func:`react_loop`.
    The provider must satisfy :class:`~executionkit.provider.ToolCallingProvider`;
    a :exc:`TypeError` is raised if it does not.
    """
    provider = self.provider
    if not _provider_supports_tools(provider):
        msg = (
            f"react() requires a ToolCallingProvider; "
            f"{type(provider).__name__} does not support tool calling."
        )
        raise TypeError(msg)
    # _provider_supports_tools verified isinstance + supports_tools=True;
    # cast is safe here because mypy cannot narrow through the helper.
    tool_provider = cast("ToolCallingProvider", provider)
    return await self._run_tracked(
        react_loop(tool_provider, prompt, tools, **kwargs)
    )

refine async

refine(prompt: str, **kwargs: Any) -> PatternResult[str]

Run the :func:~executionkit.patterns.refine_loop.refine_loop pattern.

All keyword arguments are forwarded unchanged to :func:refine_loop.

Source code in executionkit/kit.py

async def refine(self, prompt: str, **kwargs: Any) -> PatternResult[str]:
    """Run the :func:`~executionkit.patterns.refine_loop.refine_loop` pattern.

    All keyword arguments are forwarded unchanged to :func:`refine_loop`.
    """
    return await self._run_tracked(refine_loop(self.provider, prompt, **kwargs))

stream_consensus async

stream_consensus(prompt: str, *, temperature: float = _STREAM_CONSENSUS_TEMPERATURE, max_tokens: int = DEFAULT_MAX_TOKENS, max_cost: TokenUsage | None = None, trace: TraceCallback | None = None) -> StreamingPatternResult

Stream a single live completion of prompt.

Consensus voting needs complete responses to compare, so the full pattern has no coherent token stream; this convenience method streams one generation (no voting). Token deltas arrive live and result.cost becomes accurate once the stream is drained, at which point the spend is folded into this Kit's cumulative :attr:usage.

Source code in executionkit/kit.py

async def stream_consensus(
    self,
    prompt: str,
    *,
    temperature: float = _STREAM_CONSENSUS_TEMPERATURE,
    max_tokens: int = DEFAULT_MAX_TOKENS,
    max_cost: TokenUsage | None = None,
    trace: TraceCallback | None = None,
) -> StreamingPatternResult:
    """Stream a single live completion of *prompt*.

    Consensus voting needs complete responses to compare, so the full
    pattern has no coherent token stream; this convenience method streams
    **one** generation (no voting).  Token deltas arrive live and
    ``result.cost`` becomes accurate once the stream is drained, at which
    point the spend is folded into this Kit's cumulative :attr:`usage`.
    """
    return await self._stream_single(
        prompt,
        temperature=temperature,
        max_tokens=max_tokens,
        max_cost=max_cost,
        trace=trace,
    )

stream_react_loop async

stream_react_loop(prompt: str, tools: Sequence[Tool] = (), *, temperature: float = _STREAM_REACT_TEMPERATURE, max_tokens: int = DEFAULT_MAX_TOKENS, max_cost: TokenUsage | None = None, trace: TraceCallback | None = None) -> StreamingPatternResult

Stream a single live model turn for prompt.

The full ReAct loop runs tools across multiple rounds and cannot be expressed as one token stream, so this convenience method streams a single model generation. tools is accepted for parity with :meth:react but is not executed (tool-call deltas carry no message content). result.cost is accurate after the stream drains and folds into this Kit's :attr:usage.

Source code in executionkit/kit.py

async def stream_react_loop(
    self,
    prompt: str,
    tools: Sequence[Tool] = (),
    *,
    temperature: float = _STREAM_REACT_TEMPERATURE,
    max_tokens: int = DEFAULT_MAX_TOKENS,
    max_cost: TokenUsage | None = None,
    trace: TraceCallback | None = None,
) -> StreamingPatternResult:
    """Stream a single live model turn for *prompt*.

    The full ReAct loop runs tools across multiple rounds and cannot be
    expressed as one token stream, so this convenience method streams a
    single model generation.  *tools* is accepted for parity with
    :meth:`react` but is **not** executed (tool-call deltas carry no
    message content).  ``result.cost`` is accurate after the stream drains
    and folds into this Kit's :attr:`usage`.
    """
    del tools  # accepted for parity with react(); not executed when streaming
    return await self._stream_single(
        prompt,
        temperature=temperature,
        max_tokens=max_tokens,
        max_cost=max_cost,
        trace=trace,
    )

turn async

turn(user_text: str, tools: Sequence[Tool] = (), **kwargs: Any) -> PatternResult[str]

Run one conversational turn, carrying history across calls.

Appends user_text to :attr:messages, runs :func:~executionkit.patterns.react_loop.react_loop over the full history, then replaces :attr:messages with the returned transcript so the next turn continues the conversation. Cost folds into :attr:usage, exactly like :meth:react.

The provider must satisfy :class:~executionkit.provider.ToolCallingProvider; a :exc:TypeError is raised otherwise. If the turn fails (e.g. budget exhausted), :attr:messages is left unchanged so the turn can be retried.

All other keyword arguments are forwarded unchanged to react_loop.

Source code in executionkit/kit.py

async def turn(
    self,
    user_text: str,
    tools: Sequence[Tool] = (),
    **kwargs: Any,
) -> PatternResult[str]:
    """Run one conversational turn, carrying history across calls.

    Appends *user_text* to :attr:`messages`, runs
    :func:`~executionkit.patterns.react_loop.react_loop` over the full
    history, then replaces :attr:`messages` with the returned transcript so
    the next ``turn`` continues the conversation. Cost folds into
    :attr:`usage`, exactly like :meth:`react`.

    The provider must satisfy
    :class:`~executionkit.provider.ToolCallingProvider`; a :exc:`TypeError`
    is raised otherwise. If the turn fails (e.g. budget exhausted),
    :attr:`messages` is left unchanged so the turn can be retried.

    All other keyword arguments are forwarded unchanged to ``react_loop``.
    """
    provider = self.provider
    if not _provider_supports_tools(provider):
        msg = (
            f"turn() requires a ToolCallingProvider; "
            f"{type(provider).__name__} does not support tool calling."
        )
        raise TypeError(msg)
    tool_provider = cast("ToolCallingProvider", provider)
    # Build the next history without mutating self.messages until the call
    # succeeds, so a failed turn does not leave a dangling user message.
    history = [*self.messages, user_message(user_text)]
    result = await self._run_tracked(
        react_loop(tool_provider, tools=tools, messages=history, **kwargs)
    )
    transcript = result.metadata.get("messages")
    self.messages = list(transcript) if transcript is not None else history
    return result

Cost tracking

executionkit.cost.CostTracker

CostTracker()

Mutable accumulator for token and call counts.

Source code in executionkit/cost.py

def __init__(self) -> None:
    self._input: int = 0
    self._output: int = 0
    self._calls: int = 0

call_count property

call_count: int

Number of LLM calls recorded so far.

total_tokens property

total_tokens: int

Total input + output tokens recorded so far.

add_usage

add_usage(usage: TokenUsage) -> None

Add pre-computed usage to the tracker (e.g. from a pattern result).

Use this instead of accessing private fields directly.

Source code in executionkit/cost.py

def add_usage(self, usage: TokenUsage) -> None:
    """Add pre-computed usage to the tracker (e.g. from a pattern result).

    Use this instead of accessing private fields directly.
    """
    self._input += usage.input_tokens
    self._output += usage.output_tokens
    self._calls += usage.llm_calls

record

record(response: LLMResponse) -> None

Record usage from a single LLM response.

Source code in executionkit/cost.py

def record(self, response: LLMResponse) -> None:
    """Record usage from a single LLM response."""
    self._input += response.input_tokens
    self._output += response.output_tokens
    self._calls += 1

record_without_call

record_without_call(response: LLMResponse) -> None

Record token usage from a response without incrementing the call counter.

Used by :func:~executionkit.patterns.base.checked_complete because :meth:reserve_call (called from _before_attempt) pre-increments _calls before the await to uphold the asyncio concurrency contract described in the module-level docstring. Token totals are therefore recorded here, after the provider returns, without a second increment to the call counter.

Source code in executionkit/cost.py

def record_without_call(self, response: LLMResponse) -> None:
    """Record token usage from a response without incrementing the call counter.

    Used by :func:`~executionkit.patterns.base.checked_complete` because
    :meth:`reserve_call` (called from ``_before_attempt``) pre-increments
    ``_calls`` before the ``await`` to uphold the asyncio concurrency
    contract described in the module-level docstring.  Token totals are
    therefore recorded here, after the provider returns, without a second
    increment to the call counter.
    """
    self._input += response.input_tokens
    self._output += response.output_tokens

reserve_call

reserve_call() -> None

Reserve a call slot before dispatching (for budget-safe accounting).

Called by :func:~executionkit.patterns.base.checked_complete before awaiting the provider call. Reservations intentionally count every wire attempt, including failed attempts, so call budgets cap retries as well as successes.

Concurrency contract: this method must be called in the same synchronous run-segment as the preceding budget check — with no await between them — to prevent concurrent asyncio coroutines from racing past the check. See module-level docstring for details. This guarantee does NOT hold under threading.

Source code in executionkit/cost.py

def reserve_call(self) -> None:
    """Reserve a call slot before dispatching (for budget-safe accounting).

    Called by :func:`~executionkit.patterns.base.checked_complete` before
    awaiting the provider call.  Reservations intentionally count every
    wire attempt, including failed attempts, so call budgets cap retries
    as well as successes.

    **Concurrency contract**: this method must be called in the same
    synchronous run-segment as the preceding budget check — with no
    ``await`` between them — to prevent concurrent asyncio coroutines
    from racing past the check.  See module-level docstring for details.
    This guarantee does NOT hold under threading.
    """
    self._calls += 1

snapshot

snapshot() -> TokenUsage

Return current totals as an immutable snapshot without mutating state.

Equivalent to :meth:to_usage; provided as the public, intention-revealing name for callers that capture point-in-time deltas (e.g. per-step cost attribution or loop checkpoint state). Calling it never advances any counter, so two consecutive snapshots with no intervening record are equal.

Source code in executionkit/cost.py

def snapshot(self) -> TokenUsage:
    """Return current totals as an immutable snapshot without mutating state.

    Equivalent to :meth:`to_usage`; provided as the public, intention-revealing
    name for callers that capture point-in-time deltas (e.g. per-step cost
    attribution or loop checkpoint state).  Calling it never advances any
    counter, so two consecutive snapshots with no intervening ``record`` are
    equal.
    """
    return self.to_usage()

to_usage

to_usage() -> TokenUsage

Return an immutable snapshot of accumulated usage.

Source code in executionkit/cost.py

def to_usage(self) -> TokenUsage:
    """Return an immutable snapshot of accumulated usage."""
    return TokenUsage(
        input_tokens=self._input,
        output_tokens=self._output,
        llm_calls=self._calls,
    )

Engine helpers

executionkit.engine.convergence.ConvergenceDetector dataclass

ConvergenceDetector(delta_threshold: float = 0.01, patience: int = 3, score_threshold: float | None = None)

Tracks score history and detects convergence via delta + patience.

Convergence is declared when either: - score_threshold is set and the current score meets or exceeds it, or - The score delta has been below delta_threshold for patience consecutive iterations.

Attributes:

Name	Type	Description
delta_threshold	float	Minimum meaningful score improvement.
patience	int	How many consecutive stale iterations before stopping.
score_threshold	float | None	Optional absolute score target for early exit.

reset

reset() -> None

Clear all tracked state.

Source code in executionkit/engine/convergence.py

def reset(self) -> None:
    """Clear all tracked state."""
    self._scores.clear()
    self._stale_count = 0

should_stop

should_stop(score: float) -> bool

Record a score and return whether convergence is reached.

Parameters:

Name	Type	Description	Default
score	float	Evaluator score, must be in [0.0, 1.0] and not NaN.	required

Returns:

Type	Description
bool	True if the loop should stop (converged or threshold met).

Raises:

Type	Description
ValueError	If score is NaN or outside [0.0, 1.0].

Source code in executionkit/engine/convergence.py

def should_stop(self, score: float) -> bool:
    """Record a score and return whether convergence is reached.

    Args:
        score: Evaluator score, must be in [0.0, 1.0] and not NaN.

    Returns:
        True if the loop should stop (converged or threshold met).

    Raises:
        ValueError: If score is NaN or outside [0.0, 1.0].
    """
    if math.isnan(score) or not (0.0 <= score <= 1.0):
        raise ValueError(f"Invalid score: {score}")

    self._scores.append(score)

    # Absolute threshold check
    if self.score_threshold is not None and score >= self.score_threshold:
        return True

    # Delta-based convergence check (need at least 2 scores)
    if len(self._scores) >= 2:
        delta = abs(self._scores[-1] - self._scores[-2])
        if delta <= self.delta_threshold:
            self._stale_count += 1
        else:
            self._stale_count = 0

        if self._stale_count >= self.patience:
            return True

    return False

executionkit.engine.retry.RetryConfig dataclass

RetryConfig(max_retries: int = 3, base_delay: float = 1.0, max_delay: float = 60.0, exponential_base: float = 2.0, retryable: tuple[type[Exception], ...] = (RateLimitError, ProviderError), rate_limit_strategy: TokenBucket | None = None)

Immutable retry configuration with exponential backoff.

Attributes:

Name	Type	Description
max_retries	int	Maximum number of retry attempts. 0 means no retries.
base_delay	float	Base delay in seconds before first retry.
max_delay	float	Maximum delay cap in seconds.
exponential_base	float	Multiplier for exponential backoff.
retryable	tuple[type[Exception], ...]	Tuple of exception types that trigger retries.
rate_limit_strategy	TokenBucket | None	Optional TokenBucket strategy used to pace requests. When set, :func:with_retry acquires a token before each attempt and drains the bucket on RateLimitError so the provider's retry_after cooldown is honoured. None (default) preserves the original behaviour exactly.

get_delay

get_delay(attempt: int) -> float

Calculate jittered backoff delay for the given attempt (1-indexed).

Uses full jitter (random value in [0, capped_exponential]) to prevent thundering-herd effects when multiple coroutines retry simultaneously.

Source code in executionkit/engine/retry.py

def get_delay(self, attempt: int) -> float:
    """Calculate jittered backoff delay for the given attempt (1-indexed).

    Uses full jitter (random value in [0, capped_exponential]) to prevent
    thundering-herd effects when multiple coroutines retry simultaneously.
    """
    cap = min(
        self.base_delay * (self.exponential_base ** (attempt - 1)),
        self.max_delay,
    )
    return random.uniform(0.0, cap)  # noqa: S311

should_retry

should_retry(exc: Exception) -> bool

Check whether the given exception is retryable.

Source code in executionkit/engine/retry.py

def should_retry(self, exc: Exception) -> bool:
    """Check whether the given exception is retryable."""
    return isinstance(exc, self.retryable)

executionkit.engine.json_extraction.extract_json

extract_json(text: str) -> dict[str, Any] | list[Any]

Extract JSON from LLM output using multiple strategies.

Strategies (in order): 1. Raw json.loads(text.strip()) 2. Markdown code fences -- the first `json fence, then the first generic fence whose body looks like JSON. Located withstr.findso an unterminated fence in untrusted input degrades to a linear scan instead of the polynomial backtracking a.*?regex would incur under DOTALL. 3. Balanced-brace extraction -- find first{or[``, track nesting depth respecting string boundaries, find matching closer.

Parameters:

Name	Type	Description	Default
text	str	Raw LLM response text that may contain JSON.	required

Returns:

Type	Description
dict[str, Any] | list[Any]	Parsed JSON as a dict or list.

Raises:

Type	Description
ValueError	If no valid JSON can be extracted.

Source code in executionkit/engine/json_extraction.py

def extract_json(text: str) -> dict[str, Any] | list[Any]:
    """Extract JSON from LLM output using multiple strategies.

    Strategies (in order):
    1. Raw ``json.loads(text.strip())``
    2. Markdown code fences -- the first ```json fence, then the first generic
       fence whose body looks like JSON. Located with ``str.find`` so an
       unterminated fence in untrusted input degrades to a linear scan instead
       of the polynomial backtracking a ``.*?`` regex would incur under DOTALL.
    3. Balanced-brace extraction -- find first ``{`` or ``[``, track nesting
       depth respecting string boundaries, find matching closer.

    Args:
        text: Raw LLM response text that may contain JSON.

    Returns:
        Parsed JSON as a dict or list.

    Raises:
        ValueError: If no valid JSON can be extracted.
    """
    # Strategy 1: Try raw json.loads
    stripped = text.strip()
    try:
        result = json.loads(stripped)
        if isinstance(result, (dict, list)):
            return result
    except ValueError:
        pass

    # Strategy 2: Markdown code fences (linear scan, no regex backtracking)
    for candidate in _fenced_candidates(text):
        try:
            result = json.loads(candidate)
            if isinstance(result, (dict, list)):
                return result
        except ValueError:
            continue

    # Strategy 3: Balanced-brace extraction
    return _extract_balanced(text)

Errors

All exceptions inherit from ExecutionKitError and carry .cost (TokenUsage accumulated up to the failure) and .metadata (dict).

Exception	Cause
ExecutionKitError	Base for all errors.
LLMError	Base for provider communication errors.
RateLimitError	HTTP 429 — retryable. Carries retry_after.
PermanentError	HTTP 401/403/404 — not retryable.
ProviderError	Unexpected HTTP failure — retryable.
PatternError	Base for pattern logic errors.
BudgetExhaustedError	Token or call budget exceeded.
ConsensusFailedError	Unanimous strategy could not agree.
MaxIterationsError	react_loop exhausted max_rounds without a final answer.

executionkit.provider.ExecutionKitError

ExecutionKitError(message: str, *, cost: TokenUsage | None = None, metadata: dict[str, Any] | None = None)

Bases: Exception

Base exception for all ExecutionKit errors.

Source code in executionkit/errors.py

def __init__(
    self,
    message: str,
    *,
    cost: TokenUsage | None = None,
    metadata: dict[str, Any] | None = None,
) -> None:
    super().__init__(message)
    self.cost: TokenUsage = cost if cost is not None else TokenUsage()
    self.metadata: dict[str, Any] = metadata if metadata is not None else {}

executionkit.provider.RateLimitError

RateLimitError(message: str, *, retry_after: float = 1.0, cost: TokenUsage | None = None, metadata: dict[str, Any] | None = None)

Bases: LLMError

Provider returned HTTP 429 — retryable after retry_after seconds.

Source code in executionkit/errors.py

def __init__(
    self,
    message: str,
    *,
    retry_after: float = 1.0,
    cost: TokenUsage | None = None,
    metadata: dict[str, Any] | None = None,
) -> None:
    super().__init__(message, cost=cost, metadata=metadata)
    self.retry_after: float = retry_after

executionkit.provider.BudgetExhaustedError

BudgetExhaustedError(message: str, *, cost: TokenUsage | None = None, metadata: dict[str, Any] | None = None)

Bases: PatternError

Token or call budget exceeded.

Source code in executionkit/errors.py

def __init__(
    self,
    message: str,
    *,
    cost: TokenUsage | None = None,
    metadata: dict[str, Any] | None = None,
) -> None:
    super().__init__(message)
    self.cost: TokenUsage = cost if cost is not None else TokenUsage()
    self.metadata: dict[str, Any] = metadata if metadata is not None else {}

executionkit.provider.ConsensusFailedError

ConsensusFailedError(message: str, *, cost: TokenUsage | None = None, metadata: dict[str, Any] | None = None)

Bases: PatternError

Consensus pattern could not reach agreement.

Source code in executionkit/errors.py

def __init__(
    self,
    message: str,
    *,
    cost: TokenUsage | None = None,
    metadata: dict[str, Any] | None = None,
) -> None:
    super().__init__(message)
    self.cost: TokenUsage = cost if cost is not None else TokenUsage()
    self.metadata: dict[str, Any] = metadata if metadata is not None else {}

executionkit.provider.MaxIterationsError

MaxIterationsError(message: str, *, cost: TokenUsage | None = None, metadata: dict[str, Any] | None = None)

Bases: PatternError

Loop pattern exceeded its iteration limit.

Source code in executionkit/errors.py

def __init__(
    self,
    message: str,
    *,
    cost: TokenUsage | None = None,
    metadata: dict[str, Any] | None = None,
) -> None:
    super().__init__(message)
    self.cost: TokenUsage = cost if cost is not None else TokenUsage()
    self.metadata: dict[str, Any] = metadata if metadata is not None else {}