Function Execution

Function execution utilities for LLM tool calling - Version 2.

Provides structured function orchestration with clear input/output contracts and resilient error handling for the Arshai agentic framework.

Classes

class arshai.llms.utils.function_execution.FunctionCall(name, args, call_id=None, is_background=False)

Bases: object

Represents a single function call with metadata.

name: str

args: Dict[str, Any]

call_id: Optional[str] = None

is_background: bool = False

__init__(name, args, call_id=None, is_background=False)

class arshai.llms.utils.function_execution.FunctionExecutionInput(function_calls, available_functions, available_background_tasks)

Bases: object

Structured input for function execution orchestrator.

function_calls: List[FunctionCall]

available_functions: Dict[str, Callable]

available_background_tasks: Dict[str, Callable]

__init__(function_calls, available_functions, available_background_tasks)

class arshai.llms.utils.function_execution.FunctionExecutionResult(regular_results, background_initiated, failed_functions)

Bases: object

Structured result from function execution orchestrator.

regular_results: List[Dict[str, Any]]

background_initiated: List[str]

failed_functions: List[Dict[str, Any]]

__init__(regular_results, background_initiated, failed_functions)

class arshai.llms.utils.function_execution.FunctionOrchestrator

Bases: object

Simplified function execution orchestrator for LLM tool calling.

Handles parallel execution of regular functions and background tasks with resilient error handling and structured input/output.

Supports both batch execution (original) and progressive execution (new) for real-time streaming scenarios.

__init__()

async execute_functions(execution_input)

Execute function calls based on structured input and return structured results.

Parameters:

execution_input (FunctionExecutionInput) – Structured input containing function calls and available functions

Return type:

FunctionExecutionResult

Returns:

FunctionExecutionResult with results, background task status, and any errors

async execute_function_progressively(function_call, available_functions, available_background_tasks)

Execute a single function immediately and return the task.

This enables real-time progressive execution during streaming, where functions execute as soon as they’re complete rather than waiting for the entire stream to finish.

Parameters:

• function_call (FunctionCall) – Single function call to execute

• available_functions (Dict[str, Callable]) – Dict of regular callable functions

• available_background_tasks (Dict[str, Callable]) – Dict of background task callables

Return type:

Task

Returns:

asyncio.Task that can be awaited later for the result

async gather_progressive_results(function_tasks)

Gather results from progressively executed functions.

This method waits for all progressive tasks to complete and consolidates their results into a structured format.

Parameters:

function_tasks (List[Task]) – List of tasks from progressive execution

Return type:

FunctionExecutionResult

Returns:

FunctionExecutionResult with consolidated results

get_active_background_tasks_count()

Get the number of currently active background tasks.

Return type:

int

Returns:

Number of active background tasks

async wait_for_background_tasks(timeout=None)

Wait for all background tasks to complete (useful for testing).

Parameters:

timeout (float) – Maximum time to wait in seconds

Return type:

None

class arshai.llms.utils.function_execution.StreamingExecutionState(active_function_tasks=None, executed_functions=None, completed_functions=None, background_initiated=None)

Bases: object

Tracks progressive execution state during streaming.

This class maintains the state of function executions that happen progressively during streaming, preventing duplicate executions and tracking active tasks.

active_function_tasks: List[Task] = None

executed_functions: Set[str] = None

completed_functions: List[Dict[str, Any]] = None

background_initiated: List[str] = None

__post_init__()

Initialize mutable defaults properly.

add_function_task(task, function_call)

Track a new function execution.

is_already_executed(function_call)

Check if function was already executed.

Return type:

bool

clear_for_next_turn()

Clear state for the next streaming turn.

__init__(active_function_tasks=None, executed_functions=None, completed_functions=None, background_initiated=None)