core.graph

Dependency graph construction and condition evaluation.

Pure functions for building execution dependency graphs from prompt lists, determining which prompts are ready for execution, and evaluating prompt conditions.

Dependency graph construction and condition evaluation.

Pure functions for building execution dependency graphs from prompt lists, determining which prompts are ready for execution, and evaluating prompt conditions.

class DependencyEdge(from_seq, to_seq, source, condition_text=None)

Bases: object

A single dependency edge in the execution graph.

Parameters:

• from_seq (int)

• to_seq (int)

• source (str)

• condition_text (str | None)

from_seq

Sequence number of the dependency (upstream prompt).

Type:

int

to_seq

Sequence number of the dependent (downstream prompt).

Type:

int

source

How the edge was derived – ‘history’, ‘condition’, or ‘abort_condition’.

Type:

str

condition_text

The condition expression (set when source=’condition’ or ‘abort_condition’).

Type:

str | None

from_seq: int

to_seq: int

source: str

condition_text: str | None = None

class ExecutionGraph(nodes=<factory>, edges=<factory>, max_level=0)

Bases: object

Complete execution graph with dependency metadata.

Parameters:

• nodes (dict[int, PromptNode])

• edges (list[DependencyEdge])

• max_level (int)

nodes

Dictionary mapping sequence numbers to PromptNodes.

Type:

dict[int, ffai.core.prompt_node.PromptNode]

edges

List of all dependency edges with source information.

Type:

list[ffai.core.graph.DependencyEdge]

max_level

The deepest execution level in the graph.

Type:

int

nodes: dict[int, PromptNode]

edges: list[DependencyEdge]

max_level: int = 0

build_execution_graph(prompts)

Build dependency graph for parallel execution.

Delegates to build_execution_graph_with_edges and returns only the nodes dict for backward compatibility.

Parameters:

prompts (Sequence[dict[str, Any]]) – List of prompt dictionaries with sequence, prompt_name, history, and condition fields.

Returns:

Dictionary mapping sequence numbers to PromptNodes.

Raises:

ValueError – If a dependency cycle is detected.

Return type:

dict[int, PromptNode]

build_execution_graph_with_edges(prompts)

Build execution graph with full edge source metadata.

Parameters:

prompts (Sequence[dict[str, Any]]) – List of prompt dictionaries.

Returns:

ExecutionGraph with nodes, edges, and max_level.

Raises:

ValueError – If a dependency cycle is detected.

Return type:

ExecutionGraph

get_ready_prompts(state, nodes)

Get prompts ready for execution (all dependencies completed).

Parameters:

• state (ExecutionState) – Current execution state.

• nodes (dict[int, PromptNode]) – Execution graph nodes.

Returns:

List of PromptNodes ready for execution, sorted by level and sequence.

Return type:

list[PromptNode]

evaluate_condition(prompt, results_by_name, condition_field='condition')

Evaluate a prompt’s condition.

Parameters:

• prompt (dict[str, Any]) – Prompt dictionary with optional condition.

• results_by_name (dict[str, dict[str, Any]]) – Results indexed by prompt_name.

• condition_field (str) – Name of the field containing the condition expression.

Returns:

Tuple of (should_execute, condition_result, condition_error).

Return type:

tuple[bool, bool | None, str | None]

evaluate_condition_with_trace(prompt, results_by_name, condition_field='condition')

Evaluate a prompt’s condition and return the resolved trace.

Parameters:

• prompt (dict[str, Any]) – Prompt dictionary with optional condition.

• results_by_name (dict[str, dict[str, Any]]) – Results indexed by prompt_name.

• condition_field (str) – Name of the field containing the condition expression.

Returns:

Tuple of (should_execute, condition_result, condition_error, condition_trace).

Return type:

tuple[bool, bool | None, str | None, str | None]

is_abort_trigger(result)

Check whether a result triggered an abort.

Parameters:

result (dict[str, Any]) – A result dictionary from prompt execution.

Returns:

True if this result should trigger abort of remaining prompts.

Return type:

bool

Classes

class DependencyEdge(from_seq, to_seq, source, condition_text=None)

Bases: object

A single dependency edge in the execution graph.

Parameters:

• from_seq (int)

• to_seq (int)

• source (str)

• condition_text (str | None)

from_seq

Sequence number of the dependency (upstream prompt).

Type:

int

to_seq

Sequence number of the dependent (downstream prompt).

Type:

int

source

How the edge was derived – ‘history’, ‘condition’, or ‘abort_condition’.

Type:

str

condition_text

The condition expression (set when source=’condition’ or ‘abort_condition’).

Type:

str | None

from_seq: int

to_seq: int

source: str

condition_text: str | None = None

class ExecutionGraph(nodes=<factory>, edges=<factory>, max_level=0)

Bases: object

Complete execution graph with dependency metadata.

Parameters:

• nodes (dict[int, PromptNode])

• edges (list[DependencyEdge])

• max_level (int)

nodes

Dictionary mapping sequence numbers to PromptNodes.

Type:

dict[int, ffai.core.prompt_node.PromptNode]

edges

List of all dependency edges with source information.

Type:

list[ffai.core.graph.DependencyEdge]

max_level

The deepest execution level in the graph.

Type:

int

nodes: dict[int, PromptNode]

edges: list[DependencyEdge]

max_level: int = 0

Functions

build_execution_graph(prompts)

Build dependency graph for parallel execution.

Delegates to build_execution_graph_with_edges and returns only the nodes dict for backward compatibility.

Parameters:

prompts (Sequence[dict[str, Any]]) – List of prompt dictionaries with sequence, prompt_name, history, and condition fields.

Returns:

Dictionary mapping sequence numbers to PromptNodes.

Raises:

ValueError – If a dependency cycle is detected.

Return type:

dict[int, PromptNode]

build_execution_graph_with_edges(prompts)

Build execution graph with full edge source metadata.

Parameters:

prompts (Sequence[dict[str, Any]]) – List of prompt dictionaries.

Returns:

ExecutionGraph with nodes, edges, and max_level.

Raises:

ValueError – If a dependency cycle is detected.

Return type:

ExecutionGraph

evaluate_condition(prompt, results_by_name, condition_field='condition')

Evaluate a prompt’s condition.

Parameters:

• prompt (dict[str, Any]) – Prompt dictionary with optional condition.

• results_by_name (dict[str, dict[str, Any]]) – Results indexed by prompt_name.

• condition_field (str) – Name of the field containing the condition expression.

Returns:

Tuple of (should_execute, condition_result, condition_error).

Return type:

tuple[bool, bool | None, str | None]

evaluate_condition_with_trace(prompt, results_by_name, condition_field='condition')

Evaluate a prompt’s condition and return the resolved trace.

Parameters:

• prompt (dict[str, Any]) – Prompt dictionary with optional condition.

• results_by_name (dict[str, dict[str, Any]]) – Results indexed by prompt_name.

• condition_field (str) – Name of the field containing the condition expression.

Returns:

Tuple of (should_execute, condition_result, condition_error, condition_trace).

Return type:

tuple[bool, bool | None, str | None, str | None]

get_ready_prompts(state, nodes)

Get prompts ready for execution (all dependencies completed).

Parameters:

• state (ExecutionState) – Current execution state.

• nodes (dict[int, PromptNode]) – Execution graph nodes.

Returns:

List of PromptNodes ready for execution, sorted by level and sequence.

Return type:

list[PromptNode]

is_abort_trigger(result)

Check whether a result triggered an abort.

Parameters:

result (dict[str, Any]) – A result dictionary from prompt execution.

Returns:

True if this result should trigger abort of remaining prompts.

Return type:

bool