Pattern API

This page renders signatures and docstrings from the current source. The task-focused pages under Patterns explain when to use each function.

Async 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,
    # Shared across concurrent samples and forwarded by pipe().
    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. Names must be unique — tool calls are resolved by name, so a collision is rejected with ValueError before any provider call rather than shadowing the earlier tool.	()
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.
            Names must be unique — tool calls are resolved by name, so a
            collision is rejected with ``ValueError`` before any provider call
            rather than shadowing the earlier tool.
        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 = _build_tool_lookup(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),
    )

map_reduce

executionkit.patterns.map_reduce.map_reduce async

map_reduce(provider: LLMProvider, inputs: Sequence[str], *, map_prompt_template: str, reduce_prompt_template: str, max_concurrency: int = DEFAULT_MAX_CONCURRENCY, temperature: float = _DEFAULT_TEMPERATURE, max_tokens: int = DEFAULT_MAX_TOKENS, max_cost: TokenUsage | None = None, retry: RetryConfig | None = None, trace: TraceCallback | None = None, stream: bool = False) -> PatternResult[str]

Fan-out completions over inputs concurrently, then reduce.

Map phase: for each item in inputs, map_prompt_template is rendered ({item} replaced by the item) and sent to the provider. All map calls run concurrently up to max_concurrency.

Reduce phase: all map outputs are joined with a separator and substituted into reduce_prompt_template ({mapped_outputs} placeholder). A single reduce call produces the final answer.

An empty inputs sequence skips the map phase entirely and calls the reduce step with an empty string substituted for {mapped_outputs}.

Parameters:

Name	Type	Description	Default
provider	LLMProvider	LLM provider to call.	required
inputs	Sequence[str]	Items to map over. Each is substituted into map_prompt_template at {item}.	required
map_prompt_template	str	Prompt template with an {item} placeholder.	required
reduce_prompt_template	str	Prompt template with a {mapped_outputs} placeholder, which is replaced by the joined map results.	required
max_concurrency	int	Maximum simultaneous map calls. Must be >= 1.	DEFAULT_MAX_CONCURRENCY
temperature	float	Sampling temperature for both map and reduce calls.	_DEFAULT_TEMPERATURE
max_tokens	int	Maximum tokens per completion.	DEFAULT_MAX_TOKENS
max_cost	TokenUsage | None	Optional token/call budget shared across all calls.	None
retry	RetryConfig | None	Optional retry configuration per call.	None
trace	TraceCallback | None	Optional structured trace callback.	None

Returns:

Name	Type	Description
A	PatternResult[str]	class:~executionkit.types.PatternResult whose value is
	PatternResult[str]	the reduce completion, score is None, and metadata
	PatternResult[str]	includes map_count, reduce_calls, and total_calls.

Raises:

Type	Description
ValueError	If max_concurrency < 1, max_tokens < 1, or either template is missing its placeholder.
BudgetExhaustedError	If max_cost is exceeded during map or reduce.

Metadata

map_count (int): Number of items mapped (length of inputs). reduce_calls (int): Always 1 — the reduce step makes exactly one LLM call, even when inputs is empty ({mapped_outputs} is then substituted with an empty string, and the reduce call still runs). total_calls (int): Total LLM calls made — map_count + 1 when no attempt was retried. Call accounting counts every dispatched wire attempt, including failed retries (see the retry parameter), so total_calls can exceed map_count + 1 when a map or reduce call needed a retry.

Source code in executionkit/patterns/map_reduce.py

async def map_reduce(
    provider: LLMProvider,
    inputs: Sequence[str],
    *,
    map_prompt_template: str,
    reduce_prompt_template: str,
    max_concurrency: int = DEFAULT_MAX_CONCURRENCY,
    temperature: float = _DEFAULT_TEMPERATURE,
    max_tokens: int = DEFAULT_MAX_TOKENS,
    max_cost: TokenUsage | None = None,
    retry: RetryConfig | None = None,
    trace: TraceCallback | None = None,
    stream: bool = False,
) -> PatternResult[str]:
    """Fan-out completions over *inputs* concurrently, then reduce.

    **Map phase**: for each item in *inputs*, ``map_prompt_template`` is
    rendered (``{item}`` replaced by the item) and sent to the provider.
    All map calls run concurrently up to ``max_concurrency``.

    **Reduce phase**: all map outputs are joined with a separator and
    substituted into ``reduce_prompt_template`` (``{mapped_outputs}``
    placeholder).  A single reduce call produces the final answer.

    An empty *inputs* sequence skips the map phase entirely and calls the
    reduce step with an empty string substituted for ``{mapped_outputs}``.

    Args:
        provider: LLM provider to call.
        inputs: Items to map over.  Each is substituted into
            ``map_prompt_template`` at ``{item}``.
        map_prompt_template: Prompt template with an ``{item}`` placeholder.
        reduce_prompt_template: Prompt template with a ``{mapped_outputs}``
            placeholder, which is replaced by the joined map results.
        max_concurrency: Maximum simultaneous map calls.  Must be >= 1.
        temperature: Sampling temperature for both map and reduce calls.
        max_tokens: Maximum tokens per completion.
        max_cost: Optional token/call budget shared across all calls.
        retry: Optional retry configuration per call.
        trace: Optional structured trace callback.

    Returns:
        A :class:`~executionkit.types.PatternResult` whose ``value`` is
        the reduce completion, ``score`` is ``None``, and ``metadata``
        includes ``map_count``, ``reduce_calls``, and ``total_calls``.

    Raises:
        ValueError: If ``max_concurrency < 1``, ``max_tokens < 1``, or
            either template is missing its placeholder.
        BudgetExhaustedError: If ``max_cost`` is exceeded during map or
            reduce.

    Metadata:
        map_count (int): Number of items mapped (length of *inputs*).
        reduce_calls (int): Always 1 — the reduce step makes exactly one LLM
            call, even when *inputs* is empty (``{mapped_outputs}`` is then
            substituted with an empty string, and the reduce call still runs).
        total_calls (int): Total LLM calls made — ``map_count + 1`` when no
            attempt was retried. Call accounting counts every dispatched wire
            attempt, including failed retries (see the ``retry`` parameter),
            so ``total_calls`` can exceed ``map_count + 1`` when a map or
            reduce call needed a retry.
    """
    if stream:
        raise ValueError(
            "stream=True is not supported for map_reduce: this pattern "
            "aggregates intermediate results before returning."
        )
    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}")

    tracker = CostTracker()

    # -----------------------------------------------------------------
    # Map phase
    # -----------------------------------------------------------------
    map_count = len(inputs)
    mapped_outputs: list[str]

    if map_count == 0:
        mapped_outputs = []
    else:
        map_coros = [
            checked_complete(
                provider,
                [user_message(_build_map_prompt(map_prompt_template, item))],
                tracker,
                budget=max_cost,
                retry=retry,
                trace=trace,
                temperature=temperature,
                max_tokens=max_tokens,
            )
            for item in inputs
        ]
        map_responses = await gather_strict(map_coros, max_concurrency=max_concurrency)
        mapped_outputs = [r.content for r in map_responses]

    # -----------------------------------------------------------------
    # Reduce phase
    # -----------------------------------------------------------------
    reduce_prompt = _build_reduce_prompt(reduce_prompt_template, mapped_outputs)
    reduce_response = await checked_complete(
        provider,
        [user_message(reduce_prompt)],
        tracker,
        budget=max_cost,
        retry=retry,
        trace=trace,
        temperature=temperature,
        max_tokens=max_tokens,
    )

    total_calls = tracker.to_usage().llm_calls
    return PatternResult[str](
        value=reduce_response.content,
        score=None,
        cost=tracker.to_usage(),
        metadata=MappingProxyType(
            {
                "map_count": map_count,
                "reduce_calls": 1,
                "total_calls": total_calls,
            }
        ),
    )

Synchronous wrappers

The wrappers accept the same pattern-specific keyword arguments as their async counterparts. They use asyncio.run() and therefore cannot run inside an active event loop.

executionkit.consensus_sync

consensus_sync(provider: LLMProvider, prompt: str, **kwargs: Any) -> PatternResult[str]

Synchronous wrapper for :func:consensus.

Source code in executionkit/__init__.py

def consensus_sync(
    provider: LLMProvider, prompt: str, **kwargs: Any
) -> PatternResult[str]:
    """Synchronous wrapper for :func:`consensus`."""
    return _run_sync(consensus(provider, prompt, **kwargs))

executionkit.refine_loop_sync

refine_loop_sync(provider: LLMProvider, prompt: str, **kwargs: Any) -> PatternResult[str]

Synchronous wrapper for :func:refine_loop.

Source code in executionkit/__init__.py

def refine_loop_sync(
    provider: LLMProvider, prompt: str, **kwargs: Any
) -> PatternResult[str]:
    """Synchronous wrapper for :func:`refine_loop`."""
    return _run_sync(refine_loop(provider, prompt, **kwargs))

executionkit.react_loop_sync

react_loop_sync(provider: ToolCallingProvider, prompt: str, tools: Sequence[Tool] = (), **kwargs: Any) -> PatternResult[str]

Synchronous wrapper for :func:react_loop.

Source code in executionkit/__init__.py

def react_loop_sync(
    provider: ToolCallingProvider,
    prompt: str,
    tools: Sequence[Tool] = (),
    **kwargs: Any,
) -> PatternResult[str]:
    """Synchronous wrapper for :func:`react_loop`."""
    return _run_sync(react_loop(provider, prompt, tools, **kwargs))

executionkit.structured_sync

structured_sync(provider: LLMProvider, prompt: str, **kwargs: Any) -> PatternResult[Any]

Synchronous wrapper for :func:structured.

Source code in executionkit/__init__.py

def structured_sync(
    provider: LLMProvider, prompt: str, **kwargs: Any
) -> PatternResult[Any]:
    """Synchronous wrapper for :func:`structured`."""
    return _run_sync(structured(provider, prompt, **kwargs))

executionkit.pipe_sync

pipe_sync(provider: LLMProvider, prompt: str, *steps: Any, **kwargs: Any) -> PatternResult[Any]

Synchronous wrapper for :func:pipe.

Source code in executionkit/__init__.py

def pipe_sync(
    provider: LLMProvider, prompt: str, *steps: Any, **kwargs: Any
) -> PatternResult[Any]:
    """Synchronous wrapper for :func:`pipe`."""
    return _run_sync(pipe(provider, prompt, *steps, **kwargs))

executionkit.map_reduce_sync

map_reduce_sync(provider: LLMProvider, inputs: Sequence[str], **kwargs: Any) -> PatternResult[str]

Synchronous wrapper for :func:map_reduce.

Source code in executionkit/__init__.py

def map_reduce_sync(
    provider: LLMProvider,
    inputs: Sequence[str],
    **kwargs: Any,
) -> PatternResult[str]:
    """Synchronous wrapper for :func:`map_reduce`."""
    return _run_sync(map_reduce(provider, inputs, **kwargs))

Results and values

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.StreamingPatternResult dataclass

StreamingPatternResult(text_stream: AsyncIterator[str], metadata: MappingProxyType[str, Any] = (lambda: MappingProxyType({}))(), _usage_source: Callable[[], TokenUsage] = _empty_usage)

Result of a streaming pattern call — text delivered as live deltas.

Unlike :class:PatternResult, the output arrives incrementally through :attr:text_stream, an async iterator of text chunks. Token usage is only known once the provider emits its final usage frame, so :attr:cost is not meaningful until text_stream has been fully consumed; reading it earlier returns whatever has been recorded so far (typically zero).

The object itself is immutable — :attr:cost is a live view over the originating cost tracker rather than a stored snapshot, which is how a frozen result can surface usage that is populated only after the stream drains.

cost property

cost: TokenUsage

Token usage recorded so far — accurate once the stream is drained.

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.TerminationReason

Bases: StrEnum

Why a loop pattern stopped iterating.

Surfaced in :class:PatternResult metadata["termination_reason"] so callers can distinguish a clean finish from an iteration-cap truncation without catching exceptions.

Members

NATURAL: The loop reached a stopping condition before the cap (e.g. the LLM returned no tool calls in react_loop). MAX_ITERATIONS: The iteration cap was reached before the loop stopped naturally. For react_loop this means the LLM never returned a tool-call-free response within max_rounds.

Evaluator is an async callable:

Callable[[str, LLMProvider], Awaitable[float]]

It must return a finite score in the inclusive range [0.0, 1.0].

CheckpointCallback receives an iteration index and a plain state dictionary. It may be synchronous or async:

Callable[[int, dict[str, Any]], Awaitable[None] | None]

PatternStep is the callable protocol accepted by pipe():

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

Advanced helpers

Most applications should call a pattern or Kit. Custom pattern authors can use the same completion, budget, retry, and trace path as built-in patterns:

executionkit.patterns.base.checked_complete async

checked_complete(provider: LLMProvider, messages: Sequence[dict[str, Any]], tracker: CostTracker, budget: TokenUsage | None, retry: RetryConfig | None, trace: TraceCallback | None = None, **kwargs: Any) -> LLMResponse

Budget check, retry-wrapped complete(), and usage recording.

Performs three steps in order:

1. Check whether the token/call budget has been exhausted.
2. Call provider.complete() wrapped in with_retry().
3. Record the response usage on the tracker.

Concurrency model: the budget check and reserve_call() in _before_attempt are a no-await critical section — there is deliberately no await between _check_budget and tracker.reserve_call(). Under cooperative asyncio scheduling this means no other coroutine can be scheduled between the check and the reservation, so concurrent coroutines sharing the same CostTracker cannot both pass the budget check before either has incremented the counter.

This guarantee relies on cooperative scheduling (single event loop) and does NOT hold under threading. See :mod:executionkit.cost for the full concurrency contract.

Trace events carry schema-v2 payloads (see :mod:executionkit.observability): one llm_call_start per physical attempt and one terminal llm_call_end/llm_call_error per logical call, correlated by call_id.

Parameters:

Name	Type	Description	Default
provider	LLMProvider	LLM provider to call.	required
messages	Sequence[dict[str, Any]]	Chat messages to send.	required
tracker	CostTracker	Cost tracker to record usage.	required
budget	TokenUsage | None	Optional token/call budget. None means unlimited.	required
retry	RetryConfig | None	Retry configuration. Uses DEFAULT_RETRY if None.	required
trace	TraceCallback | None	Optional structured trace callback.	None
**kwargs	Any	Additional arguments forwarded to provider.complete().	{}

Returns:

Type	Description
LLMResponse	The LLM response.

Raises:

Type	Description
BudgetExhaustedError	If the budget has been exceeded before the call.
CancelledError	Always propagated (via with_retry).

Source code in executionkit/patterns/base.py

async def checked_complete(
    provider: LLMProvider,
    messages: Sequence[dict[str, Any]],
    tracker: CostTracker,
    budget: TokenUsage | None,
    retry: RetryConfig | None,
    trace: TraceCallback | None = None,
    **kwargs: Any,
) -> LLMResponse:
    """Budget check, retry-wrapped ``complete()``, and usage recording.

    Performs three steps in order:

    1. Check whether the token/call budget has been exhausted.
    2. Call ``provider.complete()`` wrapped in ``with_retry()``.
    3. Record the response usage on the tracker.

    **Concurrency model**: the budget check and ``reserve_call()`` in
    ``_before_attempt`` are a *no-await critical section* — there is
    deliberately no ``await`` between ``_check_budget`` and
    ``tracker.reserve_call()``.  Under cooperative asyncio scheduling this
    means no other coroutine can be scheduled between the check and the
    reservation, so concurrent coroutines sharing the same ``CostTracker``
    cannot both pass the budget check before either has incremented the
    counter.

    This guarantee relies on cooperative scheduling (single event loop) and
    does **NOT** hold under ``threading``.  See :mod:`executionkit.cost` for
    the full concurrency contract.

    Trace events carry schema-v2 payloads (see
    :mod:`executionkit.observability`): one ``llm_call_start`` per physical
    attempt and one terminal ``llm_call_end``/``llm_call_error`` per logical
    call, correlated by ``call_id``.

    Args:
        provider: LLM provider to call.
        messages: Chat messages to send.
        tracker: Cost tracker to record usage.
        budget: Optional token/call budget. ``None`` means unlimited.
        retry: Retry configuration. Uses ``DEFAULT_RETRY`` if ``None``.
        trace: Optional structured trace callback.
        **kwargs: Additional arguments forwarded to ``provider.complete()``.

    Returns:
        The LLM response.

    Raises:
        BudgetExhaustedError: If the budget has been exceeded before the call.
        asyncio.CancelledError: Always propagated (via ``with_retry``).
    """
    call_id = uuid4().hex
    requested_model = _requested_model(kwargs)
    started_at = _utc_now_iso()
    started_monotonic = time.monotonic()
    # Mutated by _before_attempt: the last physical attempt dispatched and its
    # id.  Plain closure state is sufficient — no other coroutine can observe
    # a partially updated pair because updates happen outside awaits.
    attempts = 0
    attempt_id: str | None = None

    async def _before_attempt(attempt: int) -> None:
        nonlocal attempts, attempt_id
        # ---- no-await critical section: check then reserve ----
        # No ``await`` must be inserted between _check_budget and
        # reserve_call().  The asyncio budget-safety guarantee depends on
        # these two operations running in the same synchronous run-segment.
        # See checked_complete() docstring and executionkit/cost.py module
        # docstring for the full concurrency contract.
        # A source-inspection test (test_no_await_between_check_and_reserve)
        # will fail CI if an await is accidentally introduced here.
        if budget is not None:
            current = tracker.to_usage()
            if attempt == 1:
                _check_budget(
                    budget,
                    current,
                    tuple(_BUDGET_FIELD_LABELS),
                    sentinel_suffix="(forwarded from pipe)",
                    exceeded_suffix="before dispatch",
                )
            else:
                _check_budget(
                    budget,
                    current,
                    ("llm_calls",),
                    sentinel_suffix="before retry (forwarded from pipe)",
                    exceeded_suffix="before retry dispatch",
                )
        tracker.reserve_call()
        # ---- end critical section ----
        attempts = attempt
        attempt_id = uuid4().hex
        await emit_trace(
            trace,
            TraceEvent.create(
                "llm_call_start",
                {
                    "schema_version": TRACE_SCHEMA_VERSION,
                    "call_id": call_id,
                    "attempt_id": attempt_id,
                    "attempt": attempt,
                    "requested_model": requested_model,
                    "started_at": _utc_now_iso(),
                    "cost": tracker.to_usage(),
                },
            ),
        )

    try:
        response = await with_retry(
            provider.complete,
            retry or DEFAULT_RETRY,
            messages,
            _before_attempt=_before_attempt,
            **kwargs,
        )
    except Exception as exc:
        await emit_trace(
            trace,
            TraceEvent.create(
                "llm_call_error",
                _error_payload(
                    exc,
                    call_id=call_id,
                    attempt_id=attempt_id,
                    attempts=attempts,
                    requested_model=requested_model,
                    started_at=started_at,
                    started_monotonic=started_monotonic,
                    tracker=tracker,
                ),
            ),
        )
        raise
    tracker.record_without_call(response)
    input_tokens, output_tokens, usage_reported = _response_usage(response)
    await emit_trace(
        trace,
        TraceEvent.create(
            "llm_call_end",
            {
                "schema_version": TRACE_SCHEMA_VERSION,
                "call_id": call_id,
                # The attempt that produced this response is the last one
                # _before_attempt dispatched.
                "attempt_id": attempt_id,
                "attempts": attempts,
                "retry_count": max(attempts - 1, 0),
                "requested_model": requested_model,
                "served_model": _served_model(response),
                "started_at": started_at,
                "duration_ms": _elapsed_ms(started_monotonic),
                "usage_reported": usage_reported,
                # None when the provider omitted usage — never a fabricated 0.
                "input_tokens": input_tokens,
                "output_tokens": output_tokens,
                "finish_reason": response.finish_reason,
                "tool_calls": _tool_call_names(response),
                "cost": tracker.to_usage(),
                # Response bodies can echo credentials supplied in the prompt;
                # redact before the library emits them so a secret in the
                # completion never leaks through the trace callback.
                "content": _redact_sensitive(response.content),
            },
        ),
    )
    return response

executionkit.patterns.base.checked_stream async

checked_stream(provider: StreamingProvider, messages: Sequence[dict[str, Any]], tracker: CostTracker, budget: TokenUsage | None, retry: RetryConfig | None, trace: TraceCallback | None = None, **kwargs: Any) -> StreamingPatternResult

Budget-checked, trace-instrumented streaming completion.

Parallel to :func:checked_complete, but yields live text deltas instead of returning a single response. Before any token is produced it eagerly:

1. Checks the budget — raises :exc:BudgetExhaustedError if exhausted.
2. Calls tracker.reserve_call() in the same no-await segment as the budget check (same concurrency contract as :func:checked_complete).
3. Emits the llm_call_start trace event.

The returned :class:StreamingPatternResult wraps the provider stream so that, once the caller drains it, token usage is recorded (record_without_call) and llm_call_end is emitted. If the stream raises mid-flight, llm_call_error is emitted and the exception propagates to the consumer.

Streaming providers commonly report no usage object at all; in that case the end event carries usage_reported: False and None token counts rather than fabricated zeros.

Retry is not supported for streaming — a partial stream cannot be replayed — so a non-None retry is logged and ignored.

Parameters:

Name	Type	Description	Default
provider	StreamingProvider	Streaming-capable LLM provider.	required
messages	Sequence[dict[str, Any]]	Chat messages to send.	required
tracker	CostTracker	Cost tracker to record usage on.	required
budget	TokenUsage | None	Optional token/call budget. None means unlimited.	required
retry	RetryConfig | None	Ignored (logged) — streaming cannot be retried.	required
trace	TraceCallback | None	Optional structured trace callback.	None
**kwargs	Any	Forwarded to provider.stream() (e.g. temperature).	{}

Returns:

Name	Type	Description
A	StreamingPatternResult	class:StreamingPatternResult whose text_stream yields deltas
	StreamingPatternResult	and whose cost becomes accurate once the stream is drained.

Raises:

Type	Description
BudgetExhaustedError	If the budget is exhausted before dispatch.

Source code in executionkit/patterns/base.py

async def checked_stream(
    provider: StreamingProvider,
    messages: Sequence[dict[str, Any]],
    tracker: CostTracker,
    budget: TokenUsage | None,
    retry: RetryConfig | None,
    trace: TraceCallback | None = None,
    **kwargs: Any,
) -> StreamingPatternResult:
    """Budget-checked, trace-instrumented streaming completion.

    Parallel to :func:`checked_complete`, but yields live text deltas instead
    of returning a single response.  Before any token is produced it eagerly:

    1. Checks the budget — raises :exc:`BudgetExhaustedError` if exhausted.
    2. Calls ``tracker.reserve_call()`` in the same no-await segment as the
       budget check (same concurrency contract as :func:`checked_complete`).
    3. Emits the ``llm_call_start`` trace event.

    The returned :class:`StreamingPatternResult` wraps the provider stream so
    that, once the caller drains it, token usage is recorded
    (``record_without_call``) and ``llm_call_end`` is emitted.  If the stream
    raises mid-flight, ``llm_call_error`` is emitted and the exception
    propagates to the consumer.

    Streaming providers commonly report **no** usage object at all; in that
    case the end event carries ``usage_reported: False`` and ``None`` token
    counts rather than fabricated zeros.

    Retry is **not** supported for streaming — a partial stream cannot be
    replayed — so a non-``None`` *retry* is logged and ignored.

    Args:
        provider: Streaming-capable LLM provider.
        messages: Chat messages to send.
        tracker: Cost tracker to record usage on.
        budget: Optional token/call budget. ``None`` means unlimited.
        retry: Ignored (logged) — streaming cannot be retried.
        trace: Optional structured trace callback.
        **kwargs: Forwarded to ``provider.stream()`` (e.g. ``temperature``).

    Returns:
        A :class:`StreamingPatternResult` whose ``text_stream`` yields deltas
        and whose ``cost`` becomes accurate once the stream is drained.

    Raises:
        BudgetExhaustedError: If the budget is exhausted before dispatch.
    """
    if retry is not None:
        logging.getLogger(__name__).warning(
            "retry is not supported for streaming and will be ignored "
            "(a partial stream cannot be replayed)."
        )

    call_id = uuid4().hex
    attempt_id = uuid4().hex
    requested_model = _requested_model(kwargs)
    started_at = _utc_now_iso()
    started_monotonic = time.monotonic()

    # ---- no-await critical section: check then reserve ----
    # No ``await`` must be inserted between _check_budget and reserve_call();
    # see checked_complete() and executionkit/cost.py for the concurrency
    # contract this upholds.  The budget refusal is *deferred* past
    # reserve_call() purely for source ordering: the AST guard in
    # tests/patterns/test_patterns_common.py asserts no ``await`` appears
    # between _check_budget and reserve_call.  On the refusal path
    # reserve_call() is skipped (budget_error is not None), so the check →
    # reserve segment stays await-free on every path that reserves.
    budget_error: BudgetExhaustedError | None = None
    if budget is not None:
        try:
            _check_budget(
                budget,
                tracker.to_usage(),
                tuple(_BUDGET_FIELD_LABELS),
                sentinel_suffix="(forwarded from pipe)",
                exceeded_suffix="before dispatch",
            )
        except BudgetExhaustedError as exc:
            budget_error = exc
    if budget_error is None:
        tracker.reserve_call()
    # ---- end critical section ----
    if budget_error is not None:
        await emit_trace(
            trace,
            TraceEvent.create(
                "llm_call_error",
                _error_payload(
                    budget_error,
                    call_id=call_id,
                    attempt_id=attempt_id,
                    attempts=0,
                    requested_model=requested_model,
                    started_at=started_at,
                    started_monotonic=started_monotonic,
                    tracker=tracker,
                    streaming=True,
                ),
            ),
        )
        raise budget_error
    await emit_trace(
        trace,
        TraceEvent.create(
            "llm_call_start",
            {
                "schema_version": TRACE_SCHEMA_VERSION,
                "call_id": call_id,
                "attempt_id": attempt_id,
                "attempt": 1,
                "requested_model": requested_model,
                "started_at": started_at,
                "streaming": True,
                "cost": tracker.to_usage(),
            },
        ),
    )

    usage_sink: list[LLMResponse] = []
    raw_stream = provider.stream(messages, usage_sink=usage_sink, **kwargs)

    async def _tracked_stream() -> AsyncIterator[str]:
        try:
            async for delta in raw_stream:
                yield delta
        except Exception as exc:
            await emit_trace(
                trace,
                TraceEvent.create(
                    "llm_call_error",
                    _error_payload(
                        exc,
                        call_id=call_id,
                        attempt_id=attempt_id,
                        attempts=1,
                        requested_model=requested_model,
                        started_at=started_at,
                        started_monotonic=started_monotonic,
                        tracker=tracker,
                        streaming=True,
                    ),
                ),
            )
            raise
        response = usage_sink[-1] if usage_sink else LLMResponse(content="")
        tracker.record_without_call(response)
        input_tokens, output_tokens, usage_reported = _response_usage(response)
        await emit_trace(
            trace,
            TraceEvent.create(
                "llm_call_end",
                {
                    "schema_version": TRACE_SCHEMA_VERSION,
                    "call_id": call_id,
                    "attempt_id": attempt_id,
                    "attempts": 1,
                    "retry_count": 0,
                    "requested_model": requested_model,
                    "served_model": _served_model(response),
                    "started_at": started_at,
                    "duration_ms": _elapsed_ms(started_monotonic),
                    "streaming": True,
                    # Streaming usage is often absent entirely; report None,
                    # never an invented zero.
                    "usage_reported": usage_reported,
                    "input_tokens": input_tokens,
                    "output_tokens": output_tokens,
                    "finish_reason": response.finish_reason,
                    "tool_calls": _tool_call_names(response),
                    "cost": tracker.to_usage(),
                    # Redact before emitting: a streamed completion can carry a
                    # credential echoed from the prompt or upstream payload.
                    "content": _redact_sensitive(response.content),
                },
            ),
        )

    return StreamingPatternResult(
        text_stream=_tracked_stream(),
        metadata=MappingProxyType({"streaming": True}),
        _usage_source=tracker.to_usage,
    )

executionkit.patterns.base.validate_score

validate_score(score: float) -> float

Validate an evaluator score is in [0.0, 1.0] and not NaN.

Parameters:

Name	Type	Description	Default
score	float	The score to validate.	required

Returns:

Type	Description
float	The validated score (unchanged).

Raises:

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

Source code in executionkit/patterns/base.py

def validate_score(score: float) -> float:
    """Validate an evaluator score is in [0.0, 1.0] and not NaN.

    Args:
        score: The score to validate.

    Returns:
        The validated score (unchanged).

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

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)

checked_complete() and checked_stream() require a caller-owned CostTracker. Preserve the source ordering of the budget check and call reservation: their no-await interval prevents concurrent asyncio tasks from passing the same call-budget slot.