core

Core FFAI infrastructure: client abstraction, prompt assembly, history, and export.

Core FFAI infrastructure: client abstraction, prompt assembly, history, and export.

class AsyncFFAIClientBase

Bases: FFAIClientBase

Async variant of FFAIClientBase for use with execute_graph.

Subclasses must implement generate_response and clone as async methods. All other methods (clear_conversation, get_conversation_history, set_conversation_history) remain synchronous because they operate on in-memory lists.

abstractmethod async generate_response(prompt, **kwargs)

Generate a response from the AI model.

Parameters:

• prompt (str) – The user prompt to send to the model.

• **kwargs (Any) – Additional model-specific parameters.

Returns:

The generated response string.

Return type:

str

abstractmethod async clone()

Create a fresh async clone of this client with empty history.

Returns:

New AsyncFFAIClientBase instance with same config, empty history.

Return type:

AsyncFFAIClientBase

class AsyncGraphExecutor(executor_fn, max_concurrency=10, prompt_resolver=None)

Bases: object

Execute a prompt DAG with topological-parallel async calls.

Parameters:

• executor_fn (Callable[..., Awaitable[ResponseResult]]) – Async callable that takes prompt kwargs and returns a ResponseResult.

• max_concurrency (int) – Maximum number of concurrent API calls, enforced by an asyncio.Semaphore.

• prompt_resolver (Callable[[dict[str, Any], dict[str, dict[str, Any]]], tuple[str, set[str]]] | None) – Optional callback (prompt_spec, results_by_name) -> (resolved_prompt, interpolated_names). When provided, each node’s prompt is resolved before execution. When None, prompts are sent as-is.

async execute(prompts)

Build and execute a prompt dependency graph.

Parameters:

prompts (list[dict[str, Any]]) – List of prompt dicts with keys: prompt_name, prompt, history, condition, abort_condition.

Returns:

GraphResult with per-prompt results and aggregate counts.

Raises:

ValueError – If a dependency cycle is detected.

Return type:

GraphResult

class ConditionEvaluator(results_by_name)

Bases: object

Safely evaluates condition expressions for conditional prompt execution.

Security model: Never uses eval() or exec() on user input. Conditions are parsed using AST and evaluated using a restricted set of operators, functions, and method calls.

Syntax:

{{prompt_name.property}} == “value” {{prompt_name.property}} != “value” {{prompt_name.property}} contains “substring” {{prompt_name.property}} not contains “substring” {{prompt_name.property}} matches “regex” len({{prompt_name.response}}) > 100 {{a.status}} == “success” and {{b.status}} == “success” {{a.status}} == “success” or {{b.status}} == “success” not {{prompt_name.has_response}}

String methods:

{{prompt_name.response}}.startswith(“prefix”) {{prompt_name.response}}.endswith(“suffix”) {{prompt_name.response}}.lower() == “value” {{prompt_name.response}}.split(“,”)[0]

JSON functions:

json_get({{prompt_name.response}}, “key.nested[0]”) json_get_default({{prompt_name.response}}, “key”, “default”) json_has({{prompt_name.response}}, “key”) json_keys({{prompt_name.response}}) “key” in json_keys({{prompt_name.response}})

Available properties:

• status: “success”, “failed”, “skipped”

• response: The AI response text

• attempts: Number of retry attempts (int)

• error: Error message if failed (str)

• has_response: True if response exists and non-empty (bool)

Note

For JSON responses, use json_has(), json_get(), or "key" in json_keys(...) to query structure. The contains and in operators always perform substring matching on the stringified response text, which can produce false matches against dict/list repr strings (e.g. "True" in {{s.response}} matches "{'pass': True}" because the four characters appear in the repr, not because of a value).

Parameters:

results_by_name (dict[str, dict[str, Any]])

ALLOWED_OPERATORS = {<class 'ast.And'>: <function ConditionEvaluator.<lambda>>, <class 'ast.Eq'>: <built-in function eq>, <class 'ast.Gt'>: <built-in function gt>, <class 'ast.GtE'>: <built-in function ge>, <class 'ast.Lt'>: <built-in function lt>, <class 'ast.LtE'>: <built-in function le>, <class 'ast.NotEq'>: <built-in function ne>, <class 'ast.Or'>: <function ConditionEvaluator.<lambda>>}

ALLOWED_UNARY_OPERATORS = {<class 'ast.Not'>: <built-in function not_>}

ALLOWED_FUNCTIONS = {'abs': <built-in function abs>, 'bool': <function ConditionEvaluator.<lambda>>, 'count': <function ConditionEvaluator.<lambda>>, 'find': <function ConditionEvaluator.<lambda>>, 'float': <function ConditionEvaluator.<lambda>>, 'int': <function ConditionEvaluator.<lambda>>, 'is_empty': <function ConditionEvaluator.<lambda>>, 'is_null': <function ConditionEvaluator.<lambda>>, 'json_get': <function ConditionEvaluator.<lambda>>, 'json_get_default': <function ConditionEvaluator.<lambda>>, 'json_has': <function ConditionEvaluator.<lambda>>, 'json_keys': <function ConditionEvaluator.<lambda>>, 'json_parse': <function ConditionEvaluator.<lambda>>, 'json_type': <function ConditionEvaluator.<lambda>>, 'json_values': <function ConditionEvaluator.<lambda>>, 'len': <built-in function len>, 'lower': <function ConditionEvaluator.<lambda>>, 'lstrip': <function ConditionEvaluator.<lambda>>, 'max': <built-in function max>, 'min': <built-in function min>, 'replace': <function ConditionEvaluator.<lambda>>, 'rfind': <function ConditionEvaluator.<lambda>>, 'round': <built-in function round>, 'rsplit': <function ConditionEvaluator.<lambda>>, 'rstrip': <function ConditionEvaluator.<lambda>>, 'slice': <function ConditionEvaluator.<lambda>>, 'split': <function ConditionEvaluator.<lambda>>, 'str': <function ConditionEvaluator.<lambda>>, 'strip': <function ConditionEvaluator.<lambda>>, 'trim': <function ConditionEvaluator.<lambda>>, 'upper': <function ConditionEvaluator.<lambda>>}

ALLOWED_STRING_METHODS = frozenset({'capitalize', 'center', 'count', 'endswith', 'find', 'index', 'isalnum', 'isalpha', 'isdecimal', 'isdigit', 'islower', 'isnumeric', 'isspace', 'istitle', 'isupper', 'join', 'ljust', 'lower', 'lstrip', 'replace', 'rfind', 'rindex', 'rjust', 'rsplit', 'rstrip', 'split', 'startswith', 'strip', 'title', 'upper', 'zfill'})

ALLOWED_LIST_METHODS = frozenset({'count', 'index'})

ALLOWED_DICT_METHODS = frozenset({'get', 'keys', 'values'})

VARIABLE_PATTERN = re.compile('\\{\\{(\\w+)\\.(\\w+)\\}\\}')

__init__(results_by_name)

Initialize evaluator with completed prompt results.

Parameters:

results_by_name (dict[str, dict[str, Any]]) – Dict mapping prompt_name to result dict with keys: - status: str - response: str - attempts: int - error: str - has_response: bool

Return type:

None

evaluate(condition)

Evaluate a condition expression.

Parameters:

condition (str) – The condition string to evaluate

Returns:

Tuple of (result, error_message) - result: True if condition passes, False otherwise - error_message: None if successful, error string if failed

Return type:

tuple[bool, str | None]

evaluate_with_trace(condition)

Evaluate a condition expression and return the resolved trace.

Like evaluate(), but also returns the resolved expression showing what each {{name.property}} was substituted with.

Parameters:

condition (str) – The condition string to evaluate.

Returns:

Tuple of (result, error_message, resolved_trace). - result: True if condition passes, False otherwise. - error_message: None if successful, error string if failed. - resolved_trace: The condition after variable substitution,

e.g. ‘“failed” == “success”’, or None if no condition.

Return type:

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

classmethod extract_referenced_names(condition)

Extract all prompt names referenced in a condition.

Parameters:

condition (str) – The condition string

Returns:

List of prompt names referenced via {{name.property}} syntax

Return type:

list

classmethod validate_syntax(condition)

Validate condition syntax without evaluating.

Parameters:

condition (str) – The condition string to validate

Returns:

Tuple of (is_valid, error_message)

Return type:

tuple[bool, str | None]

class ConversationHistory

Bases: object

API-facing message history for provider SDK calls.

Unlike PermanentHistory, turns here carry no timestamps and the history is intended to be cleared (e.g. when starting a new conversation context). The get_turns() output is structured for direct injection into provider message APIs.

add_turn_assistant(content)

Append an assistant turn to the history.

Parameters:

content (str) – The assistant’s response text.

Return type:

None

add_turn_user(content)

Append a user turn, coalescing with the previous user turn if adjacent.

Parameters:

content (str) – The user’s input text.

Return type:

None

get_turns()

Return all turns formatted for provider message APIs.

Returns:

List of message dictionaries with role and content keys.

Return type:

list[dict[str, Any]]

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 Embeddings(model='mistral/mistral-embed', api_key=None, api_base=None, cache_enabled=True, cache_size=256, device='cpu', **kwargs)

Bases: object

Compute text embeddings using local or remote models.

Supports local sentence-transformer models (prefix local/) and remote providers via LiteLLM (e.g. mistral/mistral-embed, openai/text-embedding-3-small). Results are optionally cached in an LRU-style in-memory cache.

Parameters:

• model (str) – Model identifier. Use local/<name> for local sentence-transformer models, or <provider>/<model> for remote APIs.

• api_key (str | None) – API key for the remote provider. If None, the key is read from a provider-specific environment variable.

• api_base (str | None) – Optional custom API base URL.

• cache_enabled (bool) – Whether to cache embedding results in memory.

• cache_size (int) – Maximum number of entries in the embedding cache.

• device (str) – Torch device for local models (e.g. "cpu", "cuda").

• **kwargs (Any) – Extra keyword arguments forwarded to the LiteLLM embedding call.

async aembed(texts)

Compute embeddings for one or more texts asynchronously.

Parameters:

texts (str | list[str]) – A single text string or a list of text strings.

Returns:

List of embedding vectors, one per input text.

Return type:

list[list[float]]

async aembed_single(text)

Compute the embedding for a single text asynchronously.

Parameters:

text (str) – The text to embed.

Returns:

The embedding vector.

Return type:

list[float]

embed(texts)

Compute embeddings for one or more texts synchronously.

Parameters:

texts (str | list[str]) – A single text string or a list of text strings.

Returns:

List of embedding vectors, one per input text.

Return type:

list[list[float]]

embed_single(text)

Compute the embedding for a single text synchronously.

Parameters:

text (str) – The text to embed.

Returns:

The embedding vector.

Return type:

list[float]

clear_cache()

Clear the embedding cache.

Returns:

Number of entries that were in the cache before clearing.

Return type:

int

cache_stats()

Return statistics about the embedding cache.

Returns:

Dictionary with keys enabled, max_size, and entries.

Return type:

dict[str, Any]

static cosine_similarity(a, b)

Compute cosine similarity between two vectors.

Parameters:

• a (list[float]) – First vector.

• b (list[float]) – Second vector.

Returns:

Cosine similarity score in the range [-1, 1]. Returns 0.0 if either vector has zero magnitude.

Return type:

float

property provider: str

property is_local: bool

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

class ExecutionState(completed=<factory>, in_progress=<factory>, pending=<factory>, results=<factory>, results_lock=<factory>, success_count=0, failed_count=0, skipped_count=0, aborted_count=0, results_by_name=<factory>, current_name='', aborted=False)

Bases: object

Tracks state during parallel execution with thread-safe operations.

Parameters:

• completed (set[int])

• in_progress (set[int])

• pending (dict[int, PromptNode])

• results (list[dict[str, Any]])

• results_lock (allocate_lock)

• success_count (int)

• failed_count (int)

• skipped_count (int)

• aborted_count (int)

• results_by_name (dict[str, dict[str, Any]])

• current_name (str)

• aborted (bool)

completed

Set of completed sequence numbers.

Type:

set[int]

in_progress

Set of currently executing sequence numbers.

Type:

set[int]

pending

Dict of pending PromptNodes by sequence number.

Type:

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

results

List of execution results.

Type:

list[dict[str, Any]]

results_lock

Thread lock for result access.

Type:

_thread.allocate_lock

success_count

Number of successful executions.

Type:

int

failed_count

Number of failed executions.

Type:

int

skipped_count

Number of skipped executions.

Type:

int

aborted_count

Number of aborted executions.

Type:

int

results_by_name

Results indexed by prompt name.

Type:

dict[str, dict[str, Any]]

current_name

Name of currently executing prompt.

Type:

str

aborted

Whether execution has been aborted.

Type:

bool

completed: set[int]

in_progress: set[int]

pending: dict[int, PromptNode]

results: list[dict[str, Any]]

results_lock: allocate_lock

success_count: int = 0

failed_count: int = 0

skipped_count: int = 0

aborted_count: int = 0

results_by_name: dict[str, dict[str, Any]]

current_name: str = ''

aborted: bool = False

class FFAIClientBase

Bases: ABC

Abstract base class defining the contract for AI client implementations.

All AI provider clients (Mistral, Anthropic, OpenAI, etc.) must inherit from this class and implement its abstract methods.

model

The model identifier string.

Type:

str

system_instructions

System prompt/instructions for the AI.

Type:

str

retry_config

Override retry configuration. None uses the global config defaults.

Type:

dict[str, Any] | None

model: str

system_instructions: str

retry_config: dict[str, Any] | None = None

static get_default_retry_config()

Get default retry configuration from global config.

Returns:

Dictionary with retry configuration parameters.

Return type:

dict[str, Any]

configure_retry(retry_config=None)

Configure retry behavior for this client.

Parameters:

retry_config (dict[str, Any] | None) – Optional retry configuration. If None, uses global config.

Return type:

None

property last_usage: TokenUsage | None

Token usage from the most recent generate_response() call.

property last_cost_usd: float

Estimated cost in USD from the most recent generate_response() call.

property last_duration_ms: float | None

Wall-clock duration in ms of the most recent generate_response() call.

abstractmethod generate_response(prompt, **kwargs)

Generate a response from the AI model.

Parameters:

• prompt (str) – The user prompt to send to the model.

• **kwargs (Any) – Additional model-specific parameters (temperature, max_tokens, etc.)

Returns:

The generated response string.

Return type:

str

abstractmethod clear_conversation()

Clear the conversation history.

Return type:

None

abstractmethod get_conversation_history()

Get the conversation history.

Returns:

List of message dictionaries with role and content keys.

Return type:

list[dict[str, Any]]

abstractmethod set_conversation_history(history)

Set the conversation history.

Parameters:

history (list[dict[str, Any]]) – List of message dictionaries with role and content keys.

Return type:

None

add_tool_result(tool_call_id, content)

Add a tool result to the conversation history.

Default implementation appends a tool-role message. Subclasses that manage conversation history differently (e.g. via an external API) should override this method.

Parameters:

• tool_call_id (str) – The ID of the tool call this result responds to.

• content (str) – The tool execution result string.

Return type:

None

abstractmethod clone()

Create a fresh clone of this client with empty history.

Used for thread-safe parallel execution where each thread needs an isolated client instance with the same configuration.

Returns:

New client instance with same config, empty history.

Return type:

FFAIClientBase

class GraphResult(results=<factory>, success_count=0, failed_count=0, skipped_count=0, aborted=False, aborted_count=0)

Bases: object

Aggregate result from executing a prompt dependency graph.

Parameters:

• results (dict[str, ResponseResult])

• success_count (int)

• failed_count (int)

• skipped_count (int)

• aborted (bool)

• aborted_count (int)

results

Mapping of prompt_name to ResponseResult.

Type:

dict[str, ffai.core.response_result.ResponseResult]

success_count

Number of prompts that succeeded.

Type:

int

failed_count

Number of prompts that failed.

Type:

int

skipped_count

Number of prompts skipped by conditions.

Type:

int

aborted

Whether execution was aborted.

Type:

bool

aborted_count

Number of prompts skipped due to abort.

Type:

int

results: dict[str, ResponseResult]

success_count: int = 0

failed_count: int = 0

skipped_count: int = 0

aborted: bool = False

aborted_count: int = 0

class HistoryExporter(history, clean_history, prompt_attr_history, ordered_history, persist_dir, persist_name=None, auto_persist=False)

Bases: object

DataFrame export and persistence for FFAI interaction histories.

Converts the various history stores into Polars DataFrames and handles persistence to Parquet files.

Parameters:

• history (list[dict[str, Any]]) – Raw interaction history list.

• clean_history (list[dict[str, Any]]) – Cleaned interaction history list.

• prompt_attr_history (list[dict[str, Any]]) – Prompt-attribute-indexed history list.

• ordered_history (Any) – OrderedPromptHistory instance.

• persist_dir (str) – Directory for persisted files.

• persist_name (str | None) – Base name for persisted files.

• auto_persist (bool) – Whether to auto-persist DataFrames on creation.

history_to_dataframe()

Convert the full interaction history to a polars DataFrame.

Return type:

DataFrame

clean_history_to_dataframe()

Convert the clean interaction history to a polars DataFrame.

Return type:

DataFrame

prompt_attr_history_to_dataframe()

Convert the prompt attribute history to a polars DataFrame.

Return type:

DataFrame

ordered_history_to_dataframe()

Convert the ordered interaction history to a polars DataFrame.

Return type:

DataFrame

search_history(text=None, prompt_name=None, model=None, start_time=None, end_time=None)

Search interaction history with flexible filtering options.

Parameters:

• text (str | None) – Text to search for in prompts and responses

• prompt_name (str | None) – Filter by prompt name

• model (str | None) – Filter by model name

• start_time (float | None) – Filter by timestamp (start time in epoch seconds)

• end_time (float | None) – Filter by timestamp (end time in epoch seconds)

Returns:

Filtered dataframe of interactions

Return type:

pl.DataFrame

get_model_stats_df(model_usage_stats)

Get statistics on model usage as a DataFrame.

Parameters:

model_usage_stats (dict[str, int]) – Pre-computed model usage stats dict.

Returns:

DataFrame with model usage statistics

Return type:

pl.DataFrame

get_prompt_name_stats_df(prompt_name_stats)

Get statistics on prompt name usage as a DataFrame.

Parameters:

prompt_name_stats (dict[str, int]) – Pre-computed prompt name stats dict.

Returns:

DataFrame with prompt name usage statistics

Return type:

pl.DataFrame

get_response_length_stats()

Get statistics on response lengths by prompt name.

Returns:

DataFrame with response length statistics by prompt name

Return type:

pl.DataFrame

interaction_counts_by_date()

Get counts of interactions grouped by date.

Returns:

DataFrame with interaction counts by date

Return type:

pl.DataFrame

persist_all_histories()

Persist all histories to Parquet files in the configured directory.

Return type:

bool

class Memory(embeddings, store=None)

Bases: object

Semantic recall over completed conversation turns.

Wraps an EmbeddingBackend and a TurnVectorStore. Provides two indexing entry points:

• index_turn() / aindex_turn() — extract text from a structured turn dict via turn["content"][0]["text"].

• index_turn_text() / aindex_turn_text() — embed an arbitrary caller-supplied string while still storing the structured turn dict alongside. Used by HistoryRecorder (L4) to embed the Q+A pair (f"{prompt}\n{response}") rather than just the response.

Parameters:

• embeddings (EmbeddingBackend) – Embedding backend (LiteLLM API model or local local/... model). Any object implementing EmbeddingBackend is accepted.

• store (TurnVectorStore | None) – Optional TurnVectorStore. Defaults to a fresh instance. Public and settable so callers can swap in a store loaded from Parquet (L3).

index_turn(turn, metadata=None)

Embed turn["content"][0]["text"] and store the turn.

Parameters:

• turn (dict[str, Any]) – Turn dict mirroring PermanentHistory shape. Must contain content[0]["text"].

• metadata (dict[str, Any] | None) – Optional caller metadata. Defaults to empty dict.

Returns:

The integer index of the stored entry.

Return type:

int

index_turn_text(text, turn, metadata=None)

Embed an arbitrary text and store the structured turn.

Use this when the embedded text differs from turn["content"][0]["text"] — e.g., when embedding the Q+A pair (f"{prompt}\n{response}") while still storing the response as the canonical turn content.

Parameters:

• text (str) – The plain text to embed.

• turn (dict[str, Any]) – The structured turn dict to store alongside.

• metadata (dict[str, Any] | None) – Optional caller metadata.

Returns:

The integer index of the stored entry.

Return type:

int

async aindex_turn(turn, metadata=None)

Async variant of index_turn().

Parameters:

• turn (dict[str, Any])

• metadata (dict[str, Any] | None)

Return type:

int

async aindex_turn_text(text, turn, metadata=None)

Async variant of index_turn_text().

Parameters:

• text (str)

• turn (dict[str, Any])

• metadata (dict[str, Any] | None)

Return type:

int

search(query, top_k=5, threshold=None)

Embed query and return ranked hits from the store.

Parameters:

• query (str) – Plain-text query.

• top_k (int) – Maximum hits to return.

• threshold (float | None) – Optional minimum cosine similarity.

Returns:

List of TurnHit sorted by score descending.

Return type:

list

async asearch(query, top_k=5, threshold=None)

Async variant of search().

Parameters:

• query (str)

• top_k (int)

• threshold (float | None)

Return type:

list

reindex(new_embeddings)

Re-embed all stored texts with a new embedding model.

Reads all entries via TurnVectorStore.iter_entries(), clears the store, and re-adds each entry with the new embedding. The turn dicts and metadata are preserved verbatim. After reindexing, self._embeddings is updated so subsequent search() calls embed queries with the new model.

Parameters:

new_embeddings (EmbeddingBackend) – The new embedding backend to use.

Return type:

None

count()

Number of turns currently indexed.

Return type:

int

clear()

Remove all indexed turns.

Return type:

None

class OrderedPromptHistory

Bases: object

Manages ordered prompt-response history with named references.

This class provides: - Named prompt storage for declarative context assembly - Sequential ordering of interactions - Query capabilities by prompt name, model, etc. - History chain tracking for dependency resolution

prompt_dict

OrderedDict mapping prompt names to interaction lists.

Type:

collections.OrderedDict[str, list[ffai.core.history.ordered.Interaction]]

__init__()

Initialize an empty OrderedPromptHistory.

Return type:

None

prompt_dict: OrderedDict[str, list[Interaction]]

get_effective_prompt_name(prompt_name)

Get the effective prompt name from various input types.

Parameters:

prompt_name (Any) – Can be a string, tuple, or other type.

Returns:

The cleaned prompt name as a string.

Return type:

str | tuple[str, …]

add_interaction(model, prompt, response, prompt_name=None, history=None)

Add a new interaction to the history, storing cleaned versions of prompt and response.

Parameters:

• model (str) – The model used for the interaction

• prompt (str) – The prompt text

• response (str) – The response text

• prompt_name (str | None) – Optional name/key for the prompt. If None, uses prompt text as key

• history (list[str] | None) – Optional list of prompt names that form the history chain for this interaction

Returns:

The created Interaction object

Return type:

Interaction

get_interactions_by_prompt_name(prompt_name)

Get all interactions for a specific prompt name.

Parameters:

prompt_name (str)

Return type:

list[Interaction]

get_latest_interaction_by_prompt_name(prompt_name)

Get the most recent interaction for a specific prompt name.

Parameters:

prompt_name (str)

Return type:

Interaction | None

get_all_prompt_names()

Get a list of all prompt names in order of first appearance.

Return type:

list[str]

get_all_interactions()

Get all interactions in sequence order.

Return type:

list[Interaction]

get_prompt_name_usage_stats()

Get statistics on prompt name usage.

Return type:

dict[str, int]

get_interactions_by_model_and_prompt_name(model, prompt_name)

Get all interactions for a specific model and prompt name combination.

Parameters:

• model (str)

• prompt_name (str)

Return type:

list[Interaction]

merge_histories(other)

Merge another OrderedPromptHistory into this one.

Parameters:

other (OrderedPromptHistory) – Another OrderedPromptHistory instance to merge

Return type:

None

to_dict()

Convert the entire history to a dictionary organized by prompt names.

Return type:

dict[str, list[dict[str, Any]]]

get_interaction_by_prompt(prompt)

Get an interaction by its exact prompt text.

Useful when prompt was used as the prompt_name.

Parameters:

prompt (str)

Return type:

Interaction | None

get_latest_responses_by_prompt_names(prompt_names)

Get the latest prompt and response for each specified prompt name.

Parameters:

prompt_names (list[str]) – List of prompt names to retrieve

Returns:

Dictionary with prompt names as keys and dictionaries containing ‘prompt’ and ‘response’ as values

Return type:

dict[str, dict[str, str]]

get_formatted_responses(prompt_names)

Format the latest prompts and responses including recursive history chains.

Parameters:

prompt_names (list[str]) – List of prompt names to include in the formatted output.

Returns:

Formatted string containing all prompts and responses.

Return type:

str

class PermanentHistory

Bases: object

Append-only chronological turn history with timestamps.

Each turn stores a role ("user" or "assistant"), structured content, a per-turn timestamp, and optional metadata. Consecutive user turns are coalesced by appending content unless the caller passes non-None metadata, in which case a new turn is always created (so distinct prompt_name metadata is never silently merged).

add_turn_assistant(content, metadata=None)

Append an assistant turn with the current timestamp.

Parameters:

• content (str) – The assistant’s response text.

• metadata (dict[str, Any] | None) – Optional caller metadata. When provided, stored on the turn dict under the "metadata" key. Defaults to an empty dict.

Return type:

None

add_turn_user(content, metadata=None)

Append a user turn, coalescing with the previous user turn when adjacent.

Coalescing only happens when metadata is None. If metadata is provided, a new turn is always created so distinct prompt_name metadata is never silently merged into the previous user turn.

Parameters:

• content (str) – The user’s input text.

• metadata (dict[str, Any] | None) – Optional caller metadata. When non-None, forces creation of a new turn (no coalescing).

Return type:

None

get_all_turns()

Return a deep copy of all turns with their timestamps.

Returns:

List of turn dictionaries.

Return type:

list[dict[str, Any]]

get_turns_since(timestamp)

Return turns that occurred after the specified timestamp.

Parameters:

timestamp (float) – Unix timestamp cutoff (exclusive).

Returns:

List of turn dictionaries whose timestamp is greater than timestamp.

Return type:

list[dict[str, Any]]

class PromptBuilder(prompt_attr_history)

Bases: object

Builds final prompts with history context and variable interpolation.

Reads from the shared prompt-attribute history to resolve {{prompt_name.response}} patterns and assemble conversation context blocks.

Parameters:

prompt_attr_history (list[dict[str, Any]]) – The shared prompt-attribute history list.

build_prompt(prompt, history=None, dependencies=None, strict=False)

Build the final prompt with history and variable interpolation.

Parameters:

• prompt (str) – The prompt template (may contain {{}} interpolation patterns)

• history (list[str] | None) – List of prompt names to include in conversation history

• dependencies (Any | None) – Additional dependencies (unused but kept for compatibility)

• strict (bool) – If True, raise ValueError on unknown {{name.response}} references

Returns:

Tuple of (final_prompt, set_of_interpolated_prompt_names)

Return type:

tuple[str, set[str]]

class PromptNode(sequence, prompt, dependencies=<factory>, level=0)

Bases: object

Represents a prompt in the execution dependency graph.

Parameters:

• sequence (int)

• prompt (dict[str, Any])

• dependencies (set[int])

• level (int)

sequence

The prompt’s sequence number.

Type:

int

prompt

The prompt dictionary.

Type:

dict[str, Any]

dependencies

Set of sequence numbers this prompt depends on.

Type:

set[int]

level

The execution level (0 = no dependencies).

Type:

int

sequence: int

prompt: dict[str, Any]

dependencies: set[int]

level: int = 0

is_ready(completed)

Check if this node is ready to execute.

Parameters:

completed (set[int]) – Set of completed sequence numbers.

Returns:

True if all dependencies are completed.

Return type:

bool

add_dependency(sequence)

Add a dependency to this node.

Parameters:

sequence (int) – Sequence number of the dependency.

Return type:

None

get_prompt_name()

Get the prompt name if available.

Returns:

The prompt name or None.

Return type:

str | None

class ResponseContext(shared_prompt_attr_history=None, history_lock=None)

Bases: object

Manages the shared prompt_attr_history list with thread-safe recording.

This class centralises all mutation of the shared prompt-attribute history that PromptBuilder reads from for {{name.response}} interpolation. It replaces the inline history recording previously scattered through FFAI.generate_response() and direct list appends in PlanningRunner / SynthesisRunner.

Parameters:

• shared_prompt_attr_history (list[dict[str, Any]] | None) – Optional pre-existing list to share. When None a new empty list is created.

• history_lock (threading.Lock | None) – Optional lock for thread-safe access.

record(prompt, response, model, prompt_name=None, history=None)

Record an interaction to prompt_attr_history.

If response is a dict (JSON response from the LLM), each top-level key-value pair is recorded as a separate entry so that {{key_name.response}} interpolation works.

Parameters:

• prompt (str) – The original prompt text (or JSON key name).

• response (Any) – The cleaned response (str, dict, etc.).

• model (str) – Model identifier used for this call.

• prompt_name (str | None) – Optional logical name for the prompt.

• history (list[str] | None) – Optional list of history dependency names.

Return type:

None

record_raw(interaction)

Append a pre-built interaction dict.

Used by PlanningRunner and SynthesisRunner to inject results that were not produced by FFAI.generate_response().

Parameters:

interaction (dict[str, Any]) – A dict with at least prompt, response, prompt_name keys.

Return type:

None

clear()

Clear the shared history list (used by synthesis phase reset).

Return type:

None

property history_lock: threading.Lock | None

The lock used for thread-safe history access.

class ResponseResult(response, resolved_prompt='', usage=None, cost_usd=0.0, model='', duration_ms=0.0, status='success', condition_trace=None, condition_error=None, parsed=None, parsing_errors=None)

Bases: object

Structured result from an FFAI response generation call.

Replaces side-channel attributes (last_usage, last_cost_usd, last_resolved_prompt) with a single typed return value.

Parameters:

• response (Any)

• resolved_prompt (str)

• usage (TokenUsage | None)

• cost_usd (float)

• model (str)

• duration_ms (float)

• status (str)

• condition_trace (str | None)

• condition_error (str | None)

• parsed (Any)

• parsing_errors (list[str] | None)

response

The cleaned AI response (str, dict, list, etc.).

Type:

Any

resolved_prompt

The fully interpolated prompt sent to the model.

Type:

str

usage

Token usage from the API call, if available.

Type:

ffai.core.usage.TokenUsage | None

cost_usd

Estimated cost in USD for this call.

Type:

float

model

Model identifier used for this call.

Type:

str

duration_ms

Wall-clock duration of the LLM call in milliseconds.

Type:

float

status

Execution status – “success”, “skipped”, or “failed”.

Type:

str

condition_trace

The resolved condition expression (when condition is used).

Type:

str | None

condition_error

Error message if condition evaluation failed.

Type:

str | None

parsed

Validated Pydantic model instance (when response_model is used).

Type:

Any

parsing_errors

Validation error strings if structured output parsing failed.

Type:

list[str] | None

response: Any

resolved_prompt: str = ''

usage: TokenUsage | None = None

cost_usd: float = 0.0

model: str = ''

duration_ms: float = 0.0

status: str = 'success'

condition_trace: str | None = None

condition_error: str | None = None

parsed: Any = None

parsing_errors: list[str] | None = None

class StructuredOutputHandler(max_retries=2)

Bases: object

Validates LLM responses against Pydantic models.

Parameters:

max_retries (int) – Maximum number of re-prompt attempts after validation failure. Total attempts = max_retries + 1.

build_response_format(model)

Return the Pydantic model class for use as response_format.

LiteLLM’s completion() accepts type[BaseModel] directly in the response_format parameter and handles provider-specific schema translation internally. Returning the model class instead of a hand-built dict avoids incompatible schemas across providers.

Parameters:

model (type[BaseModel]) – A Pydantic BaseModel subclass.

Returns:

The same Pydantic model class, for use as response_format.

Return type:

type[BaseModel]

build_system_suffix(model)

Generate instruction text describing the expected JSON schema.

Intended to be appended to system instructions so the model knows the exact output shape.

Parameters:

model (type[BaseModel]) – A Pydantic BaseModel subclass.

Returns:

Instruction string with embedded JSON Schema.

Return type:

str

validate(response, model)

Parse and validate an LLM response against a Pydantic model.

Uses json_repair for fault-tolerant JSON parsing, then Pydantic validation for type safety.

Parameters:

• response (str) – Raw LLM response text.

• model (type[T]) – Pydantic BaseModel subclass to validate against.

Returns:

StructuredResult with parsed set on success, or parsing_errors populated on failure.

Return type:

StructuredResult

prepare_retry_state(prompt, response_model)

Initialize retry state for a structured output loop.

Parameters:

• prompt (str) – The original resolved prompt.

• response_model (type[BaseModel]) – Pydantic BaseModel subclass.

Returns:

Tuple of (all_errors, current_prompt, best_result).

Return type:

tuple[list[str], str, StructuredResult | None]

process_attempt(response, response_model, prompt, attempt, all_errors, best_result)

Process a single structured output attempt.

Parameters:

• response (str) – The raw LLM response.

• response_model (type[BaseModel]) – Pydantic BaseModel subclass.

• prompt (str) – The original resolved prompt (for feedback).

• attempt (int) – Zero-indexed attempt number.

• all_errors (list[str]) – Accumulated error list.

• best_result (StructuredResult | None) – Best result so far.

Returns:

Tuple of (updated_best_result, next_prompt, updated_errors, should_stop).

Return type:

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

finalize_retry(best_result, all_errors, max_attempts)

Finalize after all retry attempts exhausted.

Parameters:

• best_result (StructuredResult | None) – Last validation result.

• all_errors (list[str]) – All accumulated errors.

• max_attempts (int) – Total attempts made.

Returns:

Final StructuredResult.

Return type:

StructuredResult

build_retry_feedback(errors)

Build a prompt suffix telling the LLM what went wrong.

Parameters:

errors (list[str]) – Validation error strings from the previous attempt.

Returns:

Feedback text to append to the retry prompt.

Return type:

str

class StructuredResult(parsed, raw_response, attempts, parsing_errors=<factory>)

Bases: object

Outcome of a structured-output validation attempt.

Parameters:

• parsed (BaseModel | None)

• raw_response (str)

• attempts (int)

• parsing_errors (list[str])

parsed

Validated Pydantic model instance, or None on failure.

Type:

pydantic.main.BaseModel | None

raw_response

The original LLM response string.

Type:

str

attempts

Number of validation attempts made.

Type:

int

parsing_errors

List of error strings from failed validations.

Type:

list[str]

parsed: BaseModel | None

raw_response: str

attempts: int

parsing_errors: list[str]

class TurnHit(score, turn, turn_index, text, metadata)

Bases: object

A single ranked result returned by TurnVectorStore.search.

Parameters:

• score (float)

• turn (dict[str, Any])

• turn_index (int)

• text (str)

• metadata (dict[str, Any])

score

Cosine similarity in [-1.0, 1.0] between the query embedding and the stored turn’s embedding. Sorted descending by search().

Type:

float

turn

The raw turn dict stored at index time. Mirrors the PermanentHistory turn shape: {"role": str, "content": [{"type": "text", "text": str}], "timestamp": float, "metadata": dict}.

Type:

dict[str, Any]

turn_index

Position of the entry in TurnVectorStore at the time of the search. Stable within a single search() call for a given store; may shift after clear().

Type:

int

text

Pre-extracted plain text (typically the Q+A pair) that was embedded. Equal to the string passed to add(text=...).

Type:

str

metadata

Caller-provided metadata attached at index time. Always present (possibly empty dict). Tier 2 will populate user_id / session_id / agent_id here.

Type:

dict[str, Any]

score: float

turn: dict[str, Any]

turn_index: int

text: str

metadata: dict[str, Any]

class TurnVectorStore

Bases: object

Append-only in-memory store of embedded turns with cosine search.

Entries are stored in four parallel lists keyed by an integer index. add() returns the index of the appended entry; that index is preserved on the corresponding TurnHit returned by search().

Search is O(N) over all stored entries. Adequate for Tier 1 scale; a vector-index backend (Chroma/Qdrant) is a future concern.

add(text, embedding, turn, metadata=None)

Append an entry and return its integer index.

Parameters:

• text (str) – Plain text that was embedded.

• embedding (list[float]) – Float vector returned by the embedding backend.

• turn (dict[str, Any]) – Raw turn dict (mirrors PermanentHistory turn shape).

• metadata (dict[str, Any] | None) – Caller metadata. Defaults to an empty dict.

Returns:

The integer index of the newly appended entry.

Return type:

int

search(query_embedding, top_k=5, threshold=None)

Return up to top_k turns ranked by cosine similarity.

Parameters:

• query_embedding (list[float]) – Embedded query vector.

• top_k (int) – Maximum hits to return.

• threshold (float | None) – Optional minimum cosine similarity. Hits below this score are excluded. None means no floor.

Returns:

Hits sorted by score descending. Empty list if the store is empty or all hits fall below threshold.

Return type:

list[TurnHit]

iter_entries()

Yield all entries as Entry tuples in insertion order.

Used by Memory.reindex() and persist_store() to read the store’s full contents without exposing the underlying lists.

Return type:

Iterator[Entry]

count()

Number of entries currently in the store.

Return type:

int

clear()

Remove all entries. Subsequent count() returns 0.

Return type:

None

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

build_graph_history_dict(results_by_name)

Build a {name: response_text} mapping from executor results.

Parameters:

results_by_name (dict[str, dict[str, Any]]) – Dict mapping prompt_name to result dicts with response, status, etc.

Returns:

Dict mapping prompt_name to response text (str).

Return type:

dict[str, str]

check_abort_condition(prompt_spec, results_by_name)

Check if a prompt’s abort_condition triggers a cascade abort.

Parameters:

• prompt_spec (dict[str, Any]) – Prompt dict with optional abort_condition key.

• results_by_name (dict[str, dict[str, Any]]) – Accumulated results from prior DAG levels.

Returns:

Tuple of (should_abort, trace_or_None, error_or_None).

Return type:

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

clean_response(response)

Process and validate a response, removing think tags and extracting JSON.

Parameters:

response (Any) – The raw response from the AI client.

Returns:

Cleaned response. If JSON is detected, returns the parsed object. Otherwise returns the cleaned string with think tags removed.

Return type:

Any

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]

extract_json(text)

Extract JSON from text using fault-tolerant json_repair.

Handles markdown code blocks, trailing commas, unquoted keys, and other common LLM JSON mistakes.

Parameters:

text (str) – Response text that may contain JSON.

Returns:

Parsed JSON object or None if no valid JSON found.

Return type:

Any | None

extract_json_field(data, path)

Extract a value from JSON using dot notation path.

Supports: - Simple fields: “field_name” - Nested objects: “object.field” - Array indices: “array.0” - Combined: “object.array.0.field”

Parameters:

• data (Any) – Parsed JSON data (dict or list)

• path (str) – Dot-separated path to extract

Returns:

Extracted value as string, or empty string if not found

Return type:

str

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]

interpolate_prompt(prompt, history, strict=False)

Replace {{prompt_name.response}} patterns with actual content.

Parameters:

• prompt (str) – Template containing {{}} patterns

• history (dict[str, str]) – Dict mapping prompt_name to response text

• strict (bool) – If True, raise ValueError on unknown references instead of silently replacing with empty string.

Returns:

Tuple of (resolved_prompt, set_of_interpolated_prompt_names)

Return type:

tuple[str, set[str]]

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

resolve_graph_prompt(prompt_spec, results_by_name)

Resolve interpolation and history injection for a graph node.

Performs two-phase prompt assembly:

1. Interpolation: Resolves {{name.response}} patterns using results from earlier DAG levels.

2. History injection: For each name in prompt_spec["history"], formats the prior Q&A as <conversation_history> XML. Entries already interpolated are deduplicated.

Parameters:

• prompt_spec (dict[str, Any]) – The prompt dict with keys prompt, history, etc.

• results_by_name (dict[str, dict[str, Any]]) – Accumulated results from prior DAG levels.

Returns:

Tuple of (resolved_prompt, set_of_interpolated_names).

Return type:

tuple[str, set[str]]

run_sync(coro)

Run an async coroutine from synchronous code.

If no event loop is running, delegates to asyncio.run(). If an event loop is already running (e.g. inside a Jupyter notebook or an existing async framework), runs the coroutine in a background thread to avoid RuntimeError.

Parameters:

coro (Coroutine[Any, Any, T]) – The coroutine to execute.

Returns:

The coroutine’s result.

Return type:

T

should_skip_for_failed_deps(node, results_by_name, nodes)

Check if a node should be skipped because any dependency failed.

Only failed dependencies trigger a skip. Skipped dependencies do NOT cascade — a node whose only dependency was skipped will still execute (its condition will determine whether it should proceed).

Parameters:

• node (PromptNode) – The prompt node to check.

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

• nodes (dict[int, PromptNode]) – All graph nodes indexed by sequence number.

Returns:

Tuple of (should_skip, reason_string).

Return type:

tuple[bool, str]

Classes

class AsyncFFAIClientBase

Bases: FFAIClientBase

Async variant of FFAIClientBase for use with execute_graph.

Subclasses must implement generate_response and clone as async methods. All other methods (clear_conversation, get_conversation_history, set_conversation_history) remain synchronous because they operate on in-memory lists.

abstractmethod async generate_response(prompt, **kwargs)

Generate a response from the AI model.

Parameters:

• prompt (str) – The user prompt to send to the model.

• **kwargs (Any) – Additional model-specific parameters.

Returns:

The generated response string.

Return type:

str

abstractmethod async clone()

Create a fresh async clone of this client with empty history.

Returns:

New AsyncFFAIClientBase instance with same config, empty history.

Return type:

AsyncFFAIClientBase

add_tool_result(tool_call_id, content)

Add a tool result to the conversation history.

Default implementation appends a tool-role message. Subclasses that manage conversation history differently (e.g. via an external API) should override this method.

Parameters:

• tool_call_id (str) – The ID of the tool call this result responds to.

• content (str) – The tool execution result string.

Return type:

None

abstractmethod clear_conversation()

Clear the conversation history.

Return type:

None

configure_retry(retry_config=None)

Configure retry behavior for this client.

Parameters:

retry_config (dict[str, Any] | None) – Optional retry configuration. If None, uses global config.

Return type:

None

abstractmethod get_conversation_history()

Get the conversation history.

Returns:

List of message dictionaries with role and content keys.

Return type:

list[dict[str, Any]]

static get_default_retry_config()

Get default retry configuration from global config.

Returns:

Dictionary with retry configuration parameters.

Return type:

dict[str, Any]

property last_cost_usd: float

Estimated cost in USD from the most recent generate_response() call.

property last_duration_ms: float | None

Wall-clock duration in ms of the most recent generate_response() call.

property last_usage: TokenUsage | None

Token usage from the most recent generate_response() call.

retry_config: dict[str, Any] | None = None

abstractmethod set_conversation_history(history)

Set the conversation history.

Parameters:

history (list[dict[str, Any]]) – List of message dictionaries with role and content keys.

Return type:

None

model: str

system_instructions: str

class AsyncGraphExecutor(executor_fn, max_concurrency=10, prompt_resolver=None)

Bases: object

Execute a prompt DAG with topological-parallel async calls.

Parameters:

• executor_fn (Callable[..., Awaitable[ResponseResult]]) – Async callable that takes prompt kwargs and returns a ResponseResult.

• max_concurrency (int) – Maximum number of concurrent API calls, enforced by an asyncio.Semaphore.

• prompt_resolver (Callable[[dict[str, Any], dict[str, dict[str, Any]]], tuple[str, set[str]]] | None) – Optional callback (prompt_spec, results_by_name) -> (resolved_prompt, interpolated_names). When provided, each node’s prompt is resolved before execution. When None, prompts are sent as-is.

async execute(prompts)

Build and execute a prompt dependency graph.

Parameters:

prompts (list[dict[str, Any]]) – List of prompt dicts with keys: prompt_name, prompt, history, condition, abort_condition.

Returns:

GraphResult with per-prompt results and aggregate counts.

Raises:

ValueError – If a dependency cycle is detected.

Return type:

GraphResult

class ConditionEvaluator(results_by_name)

Bases: object

Safely evaluates condition expressions for conditional prompt execution.

Security model: Never uses eval() or exec() on user input. Conditions are parsed using AST and evaluated using a restricted set of operators, functions, and method calls.

Syntax:

{{prompt_name.property}} == “value” {{prompt_name.property}} != “value” {{prompt_name.property}} contains “substring” {{prompt_name.property}} not contains “substring” {{prompt_name.property}} matches “regex” len({{prompt_name.response}}) > 100 {{a.status}} == “success” and {{b.status}} == “success” {{a.status}} == “success” or {{b.status}} == “success” not {{prompt_name.has_response}}

String methods:

{{prompt_name.response}}.startswith(“prefix”) {{prompt_name.response}}.endswith(“suffix”) {{prompt_name.response}}.lower() == “value” {{prompt_name.response}}.split(“,”)[0]

JSON functions:

json_get({{prompt_name.response}}, “key.nested[0]”) json_get_default({{prompt_name.response}}, “key”, “default”) json_has({{prompt_name.response}}, “key”) json_keys({{prompt_name.response}}) “key” in json_keys({{prompt_name.response}})

Available properties:

• status: “success”, “failed”, “skipped”

• response: The AI response text

• attempts: Number of retry attempts (int)

• error: Error message if failed (str)

• has_response: True if response exists and non-empty (bool)

Note

For JSON responses, use json_has(), json_get(), or "key" in json_keys(...) to query structure. The contains and in operators always perform substring matching on the stringified response text, which can produce false matches against dict/list repr strings (e.g. "True" in {{s.response}} matches "{'pass': True}" because the four characters appear in the repr, not because of a value).

Parameters:

results_by_name (dict[str, dict[str, Any]])

ALLOWED_OPERATORS = {<class 'ast.And'>: <function ConditionEvaluator.<lambda>>, <class 'ast.Eq'>: <built-in function eq>, <class 'ast.Gt'>: <built-in function gt>, <class 'ast.GtE'>: <built-in function ge>, <class 'ast.Lt'>: <built-in function lt>, <class 'ast.LtE'>: <built-in function le>, <class 'ast.NotEq'>: <built-in function ne>, <class 'ast.Or'>: <function ConditionEvaluator.<lambda>>}

ALLOWED_UNARY_OPERATORS = {<class 'ast.Not'>: <built-in function not_>}

ALLOWED_FUNCTIONS = {'abs': <built-in function abs>, 'bool': <function ConditionEvaluator.<lambda>>, 'count': <function ConditionEvaluator.<lambda>>, 'find': <function ConditionEvaluator.<lambda>>, 'float': <function ConditionEvaluator.<lambda>>, 'int': <function ConditionEvaluator.<lambda>>, 'is_empty': <function ConditionEvaluator.<lambda>>, 'is_null': <function ConditionEvaluator.<lambda>>, 'json_get': <function ConditionEvaluator.<lambda>>, 'json_get_default': <function ConditionEvaluator.<lambda>>, 'json_has': <function ConditionEvaluator.<lambda>>, 'json_keys': <function ConditionEvaluator.<lambda>>, 'json_parse': <function ConditionEvaluator.<lambda>>, 'json_type': <function ConditionEvaluator.<lambda>>, 'json_values': <function ConditionEvaluator.<lambda>>, 'len': <built-in function len>, 'lower': <function ConditionEvaluator.<lambda>>, 'lstrip': <function ConditionEvaluator.<lambda>>, 'max': <built-in function max>, 'min': <built-in function min>, 'replace': <function ConditionEvaluator.<lambda>>, 'rfind': <function ConditionEvaluator.<lambda>>, 'round': <built-in function round>, 'rsplit': <function ConditionEvaluator.<lambda>>, 'rstrip': <function ConditionEvaluator.<lambda>>, 'slice': <function ConditionEvaluator.<lambda>>, 'split': <function ConditionEvaluator.<lambda>>, 'str': <function ConditionEvaluator.<lambda>>, 'strip': <function ConditionEvaluator.<lambda>>, 'trim': <function ConditionEvaluator.<lambda>>, 'upper': <function ConditionEvaluator.<lambda>>}

ALLOWED_STRING_METHODS = frozenset({'capitalize', 'center', 'count', 'endswith', 'find', 'index', 'isalnum', 'isalpha', 'isdecimal', 'isdigit', 'islower', 'isnumeric', 'isspace', 'istitle', 'isupper', 'join', 'ljust', 'lower', 'lstrip', 'replace', 'rfind', 'rindex', 'rjust', 'rsplit', 'rstrip', 'split', 'startswith', 'strip', 'title', 'upper', 'zfill'})

ALLOWED_LIST_METHODS = frozenset({'count', 'index'})

ALLOWED_DICT_METHODS = frozenset({'get', 'keys', 'values'})

VARIABLE_PATTERN = re.compile('\\{\\{(\\w+)\\.(\\w+)\\}\\}')

__init__(results_by_name)

Initialize evaluator with completed prompt results.

Parameters:

results_by_name (dict[str, dict[str, Any]]) – Dict mapping prompt_name to result dict with keys: - status: str - response: str - attempts: int - error: str - has_response: bool

Return type:

None

evaluate(condition)

Evaluate a condition expression.

Parameters:

condition (str) – The condition string to evaluate

Returns:

Tuple of (result, error_message) - result: True if condition passes, False otherwise - error_message: None if successful, error string if failed

Return type:

tuple[bool, str | None]

evaluate_with_trace(condition)

Evaluate a condition expression and return the resolved trace.

Like evaluate(), but also returns the resolved expression showing what each {{name.property}} was substituted with.

Parameters:

condition (str) – The condition string to evaluate.

Returns:

Tuple of (result, error_message, resolved_trace). - result: True if condition passes, False otherwise. - error_message: None if successful, error string if failed. - resolved_trace: The condition after variable substitution,

e.g. ‘“failed” == “success”’, or None if no condition.

Return type:

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

classmethod extract_referenced_names(condition)

Extract all prompt names referenced in a condition.

Parameters:

condition (str) – The condition string

Returns:

List of prompt names referenced via {{name.property}} syntax

Return type:

list

classmethod validate_syntax(condition)

Validate condition syntax without evaluating.

Parameters:

condition (str) – The condition string to validate

Returns:

Tuple of (is_valid, error_message)

Return type:

tuple[bool, str | None]

class ConversationHistory

Bases: object

API-facing message history for provider SDK calls.

Unlike PermanentHistory, turns here carry no timestamps and the history is intended to be cleared (e.g. when starting a new conversation context). The get_turns() output is structured for direct injection into provider message APIs.

add_turn_assistant(content)

Append an assistant turn to the history.

Parameters:

content (str) – The assistant’s response text.

Return type:

None

add_turn_user(content)

Append a user turn, coalescing with the previous user turn if adjacent.

Parameters:

content (str) – The user’s input text.

Return type:

None

get_turns()

Return all turns formatted for provider message APIs.

Returns:

List of message dictionaries with role and content keys.

Return type:

list[dict[str, Any]]

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 Embeddings(model='mistral/mistral-embed', api_key=None, api_base=None, cache_enabled=True, cache_size=256, device='cpu', **kwargs)

Bases: object

Compute text embeddings using local or remote models.

Supports local sentence-transformer models (prefix local/) and remote providers via LiteLLM (e.g. mistral/mistral-embed, openai/text-embedding-3-small). Results are optionally cached in an LRU-style in-memory cache.

Parameters:

• model (str) – Model identifier. Use local/<name> for local sentence-transformer models, or <provider>/<model> for remote APIs.

• api_key (str | None) – API key for the remote provider. If None, the key is read from a provider-specific environment variable.

• api_base (str | None) – Optional custom API base URL.

• cache_enabled (bool) – Whether to cache embedding results in memory.

• cache_size (int) – Maximum number of entries in the embedding cache.

• device (str) – Torch device for local models (e.g. "cpu", "cuda").

• **kwargs (Any) – Extra keyword arguments forwarded to the LiteLLM embedding call.

async aembed(texts)

Compute embeddings for one or more texts asynchronously.

Parameters:

texts (str | list[str]) – A single text string or a list of text strings.

Returns:

List of embedding vectors, one per input text.

Return type:

list[list[float]]

async aembed_single(text)

Compute the embedding for a single text asynchronously.

Parameters:

text (str) – The text to embed.

Returns:

The embedding vector.

Return type:

list[float]

embed(texts)

Compute embeddings for one or more texts synchronously.

Parameters:

texts (str | list[str]) – A single text string or a list of text strings.

Returns:

List of embedding vectors, one per input text.

Return type:

list[list[float]]

embed_single(text)

Compute the embedding for a single text synchronously.

Parameters:

text (str) – The text to embed.

Returns:

The embedding vector.

Return type:

list[float]

clear_cache()

Clear the embedding cache.

Returns:

Number of entries that were in the cache before clearing.

Return type:

int

cache_stats()

Return statistics about the embedding cache.

Returns:

Dictionary with keys enabled, max_size, and entries.

Return type:

dict[str, Any]

static cosine_similarity(a, b)

Compute cosine similarity between two vectors.

Parameters:

• a (list[float]) – First vector.

• b (list[float]) – Second vector.

Returns:

Cosine similarity score in the range [-1, 1]. Returns 0.0 if either vector has zero magnitude.

Return type:

float

property provider: str

property is_local: bool

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

class ExecutionState(completed=<factory>, in_progress=<factory>, pending=<factory>, results=<factory>, results_lock=<factory>, success_count=0, failed_count=0, skipped_count=0, aborted_count=0, results_by_name=<factory>, current_name='', aborted=False)

Bases: object

Tracks state during parallel execution with thread-safe operations.

Parameters:

• completed (set[int])

• in_progress (set[int])

• pending (dict[int, PromptNode])

• results (list[dict[str, Any]])

• results_lock (allocate_lock)

• success_count (int)

• failed_count (int)

• skipped_count (int)

• aborted_count (int)

• results_by_name (dict[str, dict[str, Any]])

• current_name (str)

• aborted (bool)

completed

Set of completed sequence numbers.

Type:

set[int]

in_progress

Set of currently executing sequence numbers.

Type:

set[int]

pending

Dict of pending PromptNodes by sequence number.

Type:

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

results

List of execution results.

Type:

list[dict[str, Any]]

results_lock

Thread lock for result access.

Type:

_thread.allocate_lock

success_count

Number of successful executions.

Type:

int

failed_count

Number of failed executions.

Type:

int

skipped_count

Number of skipped executions.

Type:

int

aborted_count

Number of aborted executions.

Type:

int

results_by_name

Results indexed by prompt name.

Type:

dict[str, dict[str, Any]]

current_name

Name of currently executing prompt.

Type:

str

aborted

Whether execution has been aborted.

Type:

bool

completed: set[int]

in_progress: set[int]

pending: dict[int, PromptNode]

results: list[dict[str, Any]]

results_lock: allocate_lock

success_count: int = 0

failed_count: int = 0

skipped_count: int = 0

aborted_count: int = 0

results_by_name: dict[str, dict[str, Any]]

current_name: str = ''

aborted: bool = False

class FFAIClientBase

Bases: ABC

Abstract base class defining the contract for AI client implementations.

All AI provider clients (Mistral, Anthropic, OpenAI, etc.) must inherit from this class and implement its abstract methods.

model

The model identifier string.

Type:

str

system_instructions

System prompt/instructions for the AI.

Type:

str

retry_config

Override retry configuration. None uses the global config defaults.

Type:

dict[str, Any] | None

model: str

system_instructions: str

retry_config: dict[str, Any] | None = None

static get_default_retry_config()

Get default retry configuration from global config.

Returns:

Dictionary with retry configuration parameters.

Return type:

dict[str, Any]

configure_retry(retry_config=None)

Configure retry behavior for this client.

Parameters:

retry_config (dict[str, Any] | None) – Optional retry configuration. If None, uses global config.

Return type:

None

property last_usage: TokenUsage | None

Token usage from the most recent generate_response() call.

property last_cost_usd: float

Estimated cost in USD from the most recent generate_response() call.

property last_duration_ms: float | None

Wall-clock duration in ms of the most recent generate_response() call.

abstractmethod generate_response(prompt, **kwargs)

Generate a response from the AI model.

Parameters:

• prompt (str) – The user prompt to send to the model.

• **kwargs (Any) – Additional model-specific parameters (temperature, max_tokens, etc.)

Returns:

The generated response string.

Return type:

str

abstractmethod clear_conversation()

Clear the conversation history.

Return type:

None

abstractmethod get_conversation_history()

Get the conversation history.

Returns:

List of message dictionaries with role and content keys.

Return type:

list[dict[str, Any]]

abstractmethod set_conversation_history(history)

Set the conversation history.

Parameters:

history (list[dict[str, Any]]) – List of message dictionaries with role and content keys.

Return type:

None

add_tool_result(tool_call_id, content)

Add a tool result to the conversation history.

Default implementation appends a tool-role message. Subclasses that manage conversation history differently (e.g. via an external API) should override this method.

Parameters:

• tool_call_id (str) – The ID of the tool call this result responds to.

• content (str) – The tool execution result string.

Return type:

None

abstractmethod clone()

Create a fresh clone of this client with empty history.

Used for thread-safe parallel execution where each thread needs an isolated client instance with the same configuration.

Returns:

New client instance with same config, empty history.

Return type:

FFAIClientBase

class GraphResult(results=<factory>, success_count=0, failed_count=0, skipped_count=0, aborted=False, aborted_count=0)

Bases: object

Aggregate result from executing a prompt dependency graph.

Parameters:

• results (dict[str, ResponseResult])

• success_count (int)

• failed_count (int)

• skipped_count (int)

• aborted (bool)

• aborted_count (int)

results

Mapping of prompt_name to ResponseResult.

Type:

dict[str, ffai.core.response_result.ResponseResult]

success_count

Number of prompts that succeeded.

Type:

int

failed_count

Number of prompts that failed.

Type:

int

skipped_count

Number of prompts skipped by conditions.

Type:

int

aborted

Whether execution was aborted.

Type:

bool

aborted_count

Number of prompts skipped due to abort.

Type:

int

results: dict[str, ResponseResult]

success_count: int = 0

failed_count: int = 0

skipped_count: int = 0

aborted: bool = False

aborted_count: int = 0

class HistoryExporter(history, clean_history, prompt_attr_history, ordered_history, persist_dir, persist_name=None, auto_persist=False)

Bases: object

DataFrame export and persistence for FFAI interaction histories.

Converts the various history stores into Polars DataFrames and handles persistence to Parquet files.

Parameters:

• history (list[dict[str, Any]]) – Raw interaction history list.

• clean_history (list[dict[str, Any]]) – Cleaned interaction history list.

• prompt_attr_history (list[dict[str, Any]]) – Prompt-attribute-indexed history list.

• ordered_history (Any) – OrderedPromptHistory instance.

• persist_dir (str) – Directory for persisted files.

• persist_name (str | None) – Base name for persisted files.

• auto_persist (bool) – Whether to auto-persist DataFrames on creation.

history_to_dataframe()

Convert the full interaction history to a polars DataFrame.

Return type:

DataFrame

clean_history_to_dataframe()

Convert the clean interaction history to a polars DataFrame.

Return type:

DataFrame

prompt_attr_history_to_dataframe()

Convert the prompt attribute history to a polars DataFrame.

Return type:

DataFrame

ordered_history_to_dataframe()

Convert the ordered interaction history to a polars DataFrame.

Return type:

DataFrame

search_history(text=None, prompt_name=None, model=None, start_time=None, end_time=None)

Search interaction history with flexible filtering options.

Parameters:

• text (str | None) – Text to search for in prompts and responses

• prompt_name (str | None) – Filter by prompt name

• model (str | None) – Filter by model name

• start_time (float | None) – Filter by timestamp (start time in epoch seconds)

• end_time (float | None) – Filter by timestamp (end time in epoch seconds)

Returns:

Filtered dataframe of interactions

Return type:

pl.DataFrame

get_model_stats_df(model_usage_stats)

Get statistics on model usage as a DataFrame.

Parameters:

model_usage_stats (dict[str, int]) – Pre-computed model usage stats dict.

Returns:

DataFrame with model usage statistics

Return type:

pl.DataFrame

get_prompt_name_stats_df(prompt_name_stats)

Get statistics on prompt name usage as a DataFrame.

Parameters:

prompt_name_stats (dict[str, int]) – Pre-computed prompt name stats dict.

Returns:

DataFrame with prompt name usage statistics

Return type:

pl.DataFrame

get_response_length_stats()

Get statistics on response lengths by prompt name.

Returns:

DataFrame with response length statistics by prompt name

Return type:

pl.DataFrame

interaction_counts_by_date()

Get counts of interactions grouped by date.

Returns:

DataFrame with interaction counts by date

Return type:

pl.DataFrame

persist_all_histories()

Persist all histories to Parquet files in the configured directory.

Return type:

bool

class Memory(embeddings, store=None)

Bases: object

Semantic recall over completed conversation turns.

Wraps an EmbeddingBackend and a TurnVectorStore. Provides two indexing entry points:

• index_turn() / aindex_turn() — extract text from a structured turn dict via turn["content"][0]["text"].

• index_turn_text() / aindex_turn_text() — embed an arbitrary caller-supplied string while still storing the structured turn dict alongside. Used by HistoryRecorder (L4) to embed the Q+A pair (f"{prompt}\n{response}") rather than just the response.

Parameters:

• embeddings (EmbeddingBackend) – Embedding backend (LiteLLM API model or local local/... model). Any object implementing EmbeddingBackend is accepted.

• store (TurnVectorStore | None) – Optional TurnVectorStore. Defaults to a fresh instance. Public and settable so callers can swap in a store loaded from Parquet (L3).

index_turn(turn, metadata=None)

Embed turn["content"][0]["text"] and store the turn.

Parameters:

• turn (dict[str, Any]) – Turn dict mirroring PermanentHistory shape. Must contain content[0]["text"].

• metadata (dict[str, Any] | None) – Optional caller metadata. Defaults to empty dict.

Returns:

The integer index of the stored entry.

Return type:

int

index_turn_text(text, turn, metadata=None)

Embed an arbitrary text and store the structured turn.

Use this when the embedded text differs from turn["content"][0]["text"] — e.g., when embedding the Q+A pair (f"{prompt}\n{response}") while still storing the response as the canonical turn content.

Parameters:

• text (str) – The plain text to embed.

• turn (dict[str, Any]) – The structured turn dict to store alongside.

• metadata (dict[str, Any] | None) – Optional caller metadata.

Returns:

The integer index of the stored entry.

Return type:

int

async aindex_turn(turn, metadata=None)

Async variant of index_turn().

Parameters:

• turn (dict[str, Any])

• metadata (dict[str, Any] | None)

Return type:

int

async aindex_turn_text(text, turn, metadata=None)

Async variant of index_turn_text().

Parameters:

• text (str)

• turn (dict[str, Any])

• metadata (dict[str, Any] | None)

Return type:

int

search(query, top_k=5, threshold=None)

Embed query and return ranked hits from the store.

Parameters:

• query (str) – Plain-text query.

• top_k (int) – Maximum hits to return.

• threshold (float | None) – Optional minimum cosine similarity.

Returns:

List of TurnHit sorted by score descending.

Return type:

list

async asearch(query, top_k=5, threshold=None)

Async variant of search().

Parameters:

• query (str)

• top_k (int)

• threshold (float | None)

Return type:

list

reindex(new_embeddings)

Re-embed all stored texts with a new embedding model.

Reads all entries via TurnVectorStore.iter_entries(), clears the store, and re-adds each entry with the new embedding. The turn dicts and metadata are preserved verbatim. After reindexing, self._embeddings is updated so subsequent search() calls embed queries with the new model.

Parameters:

new_embeddings (EmbeddingBackend) – The new embedding backend to use.

Return type:

None

count()

Number of turns currently indexed.

Return type:

int

clear()

Remove all indexed turns.

Return type:

None

class OrderedPromptHistory

Bases: object

Manages ordered prompt-response history with named references.

This class provides: - Named prompt storage for declarative context assembly - Sequential ordering of interactions - Query capabilities by prompt name, model, etc. - History chain tracking for dependency resolution

prompt_dict

OrderedDict mapping prompt names to interaction lists.

Type:

collections.OrderedDict[str, list[ffai.core.history.ordered.Interaction]]

__init__()

Initialize an empty OrderedPromptHistory.

Return type:

None

prompt_dict: OrderedDict[str, list[Interaction]]

get_effective_prompt_name(prompt_name)

Get the effective prompt name from various input types.

Parameters:

prompt_name (Any) – Can be a string, tuple, or other type.

Returns:

The cleaned prompt name as a string.

Return type:

str | tuple[str, …]

add_interaction(model, prompt, response, prompt_name=None, history=None)

Add a new interaction to the history, storing cleaned versions of prompt and response.

Parameters:

• model (str) – The model used for the interaction

• prompt (str) – The prompt text

• response (str) – The response text

• prompt_name (str | None) – Optional name/key for the prompt. If None, uses prompt text as key

• history (list[str] | None) – Optional list of prompt names that form the history chain for this interaction

Returns:

The created Interaction object

Return type:

Interaction

get_interactions_by_prompt_name(prompt_name)

Get all interactions for a specific prompt name.

Parameters:

prompt_name (str)

Return type:

list[Interaction]

get_latest_interaction_by_prompt_name(prompt_name)

Get the most recent interaction for a specific prompt name.

Parameters:

prompt_name (str)

Return type:

Interaction | None

get_all_prompt_names()

Get a list of all prompt names in order of first appearance.

Return type:

list[str]

get_all_interactions()

Get all interactions in sequence order.

Return type:

list[Interaction]

get_prompt_name_usage_stats()

Get statistics on prompt name usage.

Return type:

dict[str, int]

get_interactions_by_model_and_prompt_name(model, prompt_name)

Get all interactions for a specific model and prompt name combination.

Parameters:

• model (str)

• prompt_name (str)

Return type:

list[Interaction]

merge_histories(other)

Merge another OrderedPromptHistory into this one.

Parameters:

other (OrderedPromptHistory) – Another OrderedPromptHistory instance to merge

Return type:

None

to_dict()

Convert the entire history to a dictionary organized by prompt names.

Return type:

dict[str, list[dict[str, Any]]]

get_interaction_by_prompt(prompt)

Get an interaction by its exact prompt text.

Useful when prompt was used as the prompt_name.

Parameters:

prompt (str)

Return type:

Interaction | None

get_latest_responses_by_prompt_names(prompt_names)

Get the latest prompt and response for each specified prompt name.

Parameters:

prompt_names (list[str]) – List of prompt names to retrieve

Returns:

Dictionary with prompt names as keys and dictionaries containing ‘prompt’ and ‘response’ as values

Return type:

dict[str, dict[str, str]]

get_formatted_responses(prompt_names)

Format the latest prompts and responses including recursive history chains.

Parameters:

prompt_names (list[str]) – List of prompt names to include in the formatted output.

Returns:

Formatted string containing all prompts and responses.

Return type:

str

class PermanentHistory

Bases: object

Append-only chronological turn history with timestamps.

Each turn stores a role ("user" or "assistant"), structured content, a per-turn timestamp, and optional metadata. Consecutive user turns are coalesced by appending content unless the caller passes non-None metadata, in which case a new turn is always created (so distinct prompt_name metadata is never silently merged).

add_turn_assistant(content, metadata=None)

Append an assistant turn with the current timestamp.

Parameters:

• content (str) – The assistant’s response text.

• metadata (dict[str, Any] | None) – Optional caller metadata. When provided, stored on the turn dict under the "metadata" key. Defaults to an empty dict.

Return type:

None

add_turn_user(content, metadata=None)

Append a user turn, coalescing with the previous user turn when adjacent.

Coalescing only happens when metadata is None. If metadata is provided, a new turn is always created so distinct prompt_name metadata is never silently merged into the previous user turn.

Parameters:

• content (str) – The user’s input text.

• metadata (dict[str, Any] | None) – Optional caller metadata. When non-None, forces creation of a new turn (no coalescing).

Return type:

None

get_all_turns()

Return a deep copy of all turns with their timestamps.

Returns:

List of turn dictionaries.

Return type:

list[dict[str, Any]]

get_turns_since(timestamp)

Return turns that occurred after the specified timestamp.

Parameters:

timestamp (float) – Unix timestamp cutoff (exclusive).

Returns:

List of turn dictionaries whose timestamp is greater than timestamp.

Return type:

list[dict[str, Any]]

class PromptBuilder(prompt_attr_history)

Bases: object

Builds final prompts with history context and variable interpolation.

Reads from the shared prompt-attribute history to resolve {{prompt_name.response}} patterns and assemble conversation context blocks.

Parameters:

prompt_attr_history (list[dict[str, Any]]) – The shared prompt-attribute history list.

build_prompt(prompt, history=None, dependencies=None, strict=False)

Build the final prompt with history and variable interpolation.

Parameters:

• prompt (str) – The prompt template (may contain {{}} interpolation patterns)

• history (list[str] | None) – List of prompt names to include in conversation history

• dependencies (Any | None) – Additional dependencies (unused but kept for compatibility)

• strict (bool) – If True, raise ValueError on unknown {{name.response}} references

Returns:

Tuple of (final_prompt, set_of_interpolated_prompt_names)

Return type:

tuple[str, set[str]]

class PromptNode(sequence, prompt, dependencies=<factory>, level=0)

Bases: object

Represents a prompt in the execution dependency graph.

Parameters:

• sequence (int)

• prompt (dict[str, Any])

• dependencies (set[int])

• level (int)

sequence

The prompt’s sequence number.

Type:

int

prompt

The prompt dictionary.

Type:

dict[str, Any]

dependencies

Set of sequence numbers this prompt depends on.

Type:

set[int]

level

The execution level (0 = no dependencies).

Type:

int

sequence: int

prompt: dict[str, Any]

dependencies: set[int]

level: int = 0

is_ready(completed)

Check if this node is ready to execute.

Parameters:

completed (set[int]) – Set of completed sequence numbers.

Returns:

True if all dependencies are completed.

Return type:

bool

add_dependency(sequence)

Add a dependency to this node.

Parameters:

sequence (int) – Sequence number of the dependency.

Return type:

None

get_prompt_name()

Get the prompt name if available.

Returns:

The prompt name or None.

Return type:

str | None

class ResponseContext(shared_prompt_attr_history=None, history_lock=None)

Bases: object

Manages the shared prompt_attr_history list with thread-safe recording.

This class centralises all mutation of the shared prompt-attribute history that PromptBuilder reads from for {{name.response}} interpolation. It replaces the inline history recording previously scattered through FFAI.generate_response() and direct list appends in PlanningRunner / SynthesisRunner.

Parameters:

• shared_prompt_attr_history (list[dict[str, Any]] | None) – Optional pre-existing list to share. When None a new empty list is created.

• history_lock (threading.Lock | None) – Optional lock for thread-safe access.

record(prompt, response, model, prompt_name=None, history=None)

Record an interaction to prompt_attr_history.

If response is a dict (JSON response from the LLM), each top-level key-value pair is recorded as a separate entry so that {{key_name.response}} interpolation works.

Parameters:

• prompt (str) – The original prompt text (or JSON key name).

• response (Any) – The cleaned response (str, dict, etc.).

• model (str) – Model identifier used for this call.

• prompt_name (str | None) – Optional logical name for the prompt.

• history (list[str] | None) – Optional list of history dependency names.

Return type:

None

record_raw(interaction)

Append a pre-built interaction dict.

Used by PlanningRunner and SynthesisRunner to inject results that were not produced by FFAI.generate_response().

Parameters:

interaction (dict[str, Any]) – A dict with at least prompt, response, prompt_name keys.

Return type:

None

clear()

Clear the shared history list (used by synthesis phase reset).

Return type:

None

property history_lock: threading.Lock | None

The lock used for thread-safe history access.

class ResponseResult(response, resolved_prompt='', usage=None, cost_usd=0.0, model='', duration_ms=0.0, status='success', condition_trace=None, condition_error=None, parsed=None, parsing_errors=None)

Bases: object

Structured result from an FFAI response generation call.

Replaces side-channel attributes (last_usage, last_cost_usd, last_resolved_prompt) with a single typed return value.

Parameters:

• response (Any)

• resolved_prompt (str)

• usage (TokenUsage | None)

• cost_usd (float)

• model (str)

• duration_ms (float)

• status (str)

• condition_trace (str | None)

• condition_error (str | None)

• parsed (Any)

• parsing_errors (list[str] | None)

response

The cleaned AI response (str, dict, list, etc.).

Type:

Any

resolved_prompt

The fully interpolated prompt sent to the model.

Type:

str

usage

Token usage from the API call, if available.

Type:

ffai.core.usage.TokenUsage | None

cost_usd

Estimated cost in USD for this call.

Type:

float

model

Model identifier used for this call.

Type:

str

duration_ms

Wall-clock duration of the LLM call in milliseconds.

Type:

float

status

Execution status – “success”, “skipped”, or “failed”.

Type:

str

condition_trace

The resolved condition expression (when condition is used).

Type:

str | None

condition_error

Error message if condition evaluation failed.

Type:

str | None

parsed

Validated Pydantic model instance (when response_model is used).

Type:

Any

parsing_errors

Validation error strings if structured output parsing failed.

Type:

list[str] | None

response: Any

resolved_prompt: str = ''

usage: TokenUsage | None = None

cost_usd: float = 0.0

model: str = ''

duration_ms: float = 0.0

status: str = 'success'

condition_trace: str | None = None

condition_error: str | None = None

parsed: Any = None

parsing_errors: list[str] | None = None

class StructuredOutputHandler(max_retries=2)

Bases: object

Validates LLM responses against Pydantic models.

Parameters:

max_retries (int) – Maximum number of re-prompt attempts after validation failure. Total attempts = max_retries + 1.

build_response_format(model)

Return the Pydantic model class for use as response_format.

LiteLLM’s completion() accepts type[BaseModel] directly in the response_format parameter and handles provider-specific schema translation internally. Returning the model class instead of a hand-built dict avoids incompatible schemas across providers.

Parameters:

model (type[BaseModel]) – A Pydantic BaseModel subclass.

Returns:

The same Pydantic model class, for use as response_format.

Return type:

type[BaseModel]

build_system_suffix(model)

Generate instruction text describing the expected JSON schema.

Intended to be appended to system instructions so the model knows the exact output shape.

Parameters:

model (type[BaseModel]) – A Pydantic BaseModel subclass.

Returns:

Instruction string with embedded JSON Schema.

Return type:

str

validate(response, model)

Parse and validate an LLM response against a Pydantic model.

Uses json_repair for fault-tolerant JSON parsing, then Pydantic validation for type safety.

Parameters:

• response (str) – Raw LLM response text.

• model (type[T]) – Pydantic BaseModel subclass to validate against.

Returns:

StructuredResult with parsed set on success, or parsing_errors populated on failure.

Return type:

StructuredResult

prepare_retry_state(prompt, response_model)

Initialize retry state for a structured output loop.

Parameters:

• prompt (str) – The original resolved prompt.

• response_model (type[BaseModel]) – Pydantic BaseModel subclass.

Returns:

Tuple of (all_errors, current_prompt, best_result).

Return type:

tuple[list[str], str, StructuredResult | None]

process_attempt(response, response_model, prompt, attempt, all_errors, best_result)

Process a single structured output attempt.

Parameters:

• response (str) – The raw LLM response.

• response_model (type[BaseModel]) – Pydantic BaseModel subclass.

• prompt (str) – The original resolved prompt (for feedback).

• attempt (int) – Zero-indexed attempt number.

• all_errors (list[str]) – Accumulated error list.

• best_result (StructuredResult | None) – Best result so far.

Returns:

Tuple of (updated_best_result, next_prompt, updated_errors, should_stop).

Return type:

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

finalize_retry(best_result, all_errors, max_attempts)

Finalize after all retry attempts exhausted.

Parameters:

• best_result (StructuredResult | None) – Last validation result.

• all_errors (list[str]) – All accumulated errors.

• max_attempts (int) – Total attempts made.

Returns:

Final StructuredResult.

Return type:

StructuredResult

build_retry_feedback(errors)

Build a prompt suffix telling the LLM what went wrong.

Parameters:

errors (list[str]) – Validation error strings from the previous attempt.

Returns:

Feedback text to append to the retry prompt.

Return type:

str

class StructuredResult(parsed, raw_response, attempts, parsing_errors=<factory>)

Bases: object

Outcome of a structured-output validation attempt.

Parameters:

• parsed (BaseModel | None)

• raw_response (str)

• attempts (int)

• parsing_errors (list[str])

parsed

Validated Pydantic model instance, or None on failure.

Type:

pydantic.main.BaseModel | None

raw_response

The original LLM response string.

Type:

str

attempts

Number of validation attempts made.

Type:

int

parsing_errors

List of error strings from failed validations.

Type:

list[str]

parsed: BaseModel | None

raw_response: str

attempts: int

parsing_errors: list[str]

class TurnHit(score, turn, turn_index, text, metadata)

Bases: object

A single ranked result returned by TurnVectorStore.search.

Parameters:

• score (float)

• turn (dict[str, Any])

• turn_index (int)

• text (str)

• metadata (dict[str, Any])

score

Cosine similarity in [-1.0, 1.0] between the query embedding and the stored turn’s embedding. Sorted descending by search().

Type:

float

turn

The raw turn dict stored at index time. Mirrors the PermanentHistory turn shape: {"role": str, "content": [{"type": "text", "text": str}], "timestamp": float, "metadata": dict}.

Type:

dict[str, Any]

turn_index

Position of the entry in TurnVectorStore at the time of the search. Stable within a single search() call for a given store; may shift after clear().

Type:

int

text

Pre-extracted plain text (typically the Q+A pair) that was embedded. Equal to the string passed to add(text=...).

Type:

str

metadata

Caller-provided metadata attached at index time. Always present (possibly empty dict). Tier 2 will populate user_id / session_id / agent_id here.

Type:

dict[str, Any]

score: float

turn: dict[str, Any]

turn_index: int

text: str

metadata: dict[str, Any]

class TurnVectorStore

Bases: object

Append-only in-memory store of embedded turns with cosine search.

Entries are stored in four parallel lists keyed by an integer index. add() returns the index of the appended entry; that index is preserved on the corresponding TurnHit returned by search().

Search is O(N) over all stored entries. Adequate for Tier 1 scale; a vector-index backend (Chroma/Qdrant) is a future concern.

add(text, embedding, turn, metadata=None)

Append an entry and return its integer index.

Parameters:

• text (str) – Plain text that was embedded.

• embedding (list[float]) – Float vector returned by the embedding backend.

• turn (dict[str, Any]) – Raw turn dict (mirrors PermanentHistory turn shape).

• metadata (dict[str, Any] | None) – Caller metadata. Defaults to an empty dict.

Returns:

The integer index of the newly appended entry.

Return type:

int

search(query_embedding, top_k=5, threshold=None)

Return up to top_k turns ranked by cosine similarity.

Parameters:

• query_embedding (list[float]) – Embedded query vector.

• top_k (int) – Maximum hits to return.

• threshold (float | None) – Optional minimum cosine similarity. Hits below this score are excluded. None means no floor.

Returns:

Hits sorted by score descending. Empty list if the store is empty or all hits fall below threshold.

Return type:

list[TurnHit]

iter_entries()

Yield all entries as Entry tuples in insertion order.

Used by Memory.reindex() and persist_store() to read the store’s full contents without exposing the underlying lists.

Return type:

Iterator[Entry]

count()

Number of entries currently in the store.

Return type:

int

clear()

Remove all entries. Subsequent count() returns 0.

Return type:

None

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

build_graph_history_dict(results_by_name)

Build a {name: response_text} mapping from executor results.

Parameters:

results_by_name (dict[str, dict[str, Any]]) – Dict mapping prompt_name to result dicts with response, status, etc.

Returns:

Dict mapping prompt_name to response text (str).

Return type:

dict[str, str]

check_abort_condition(prompt_spec, results_by_name)

Check if a prompt’s abort_condition triggers a cascade abort.

Parameters:

• prompt_spec (dict[str, Any]) – Prompt dict with optional abort_condition key.

• results_by_name (dict[str, dict[str, Any]]) – Accumulated results from prior DAG levels.

Returns:

Tuple of (should_abort, trace_or_None, error_or_None).

Return type:

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

clean_response(response)

Process and validate a response, removing think tags and extracting JSON.

Parameters:

response (Any) – The raw response from the AI client.

Returns:

Cleaned response. If JSON is detected, returns the parsed object. Otherwise returns the cleaned string with think tags removed.

Return type:

Any

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]

extract_json(text)

Extract JSON from text using fault-tolerant json_repair.

Handles markdown code blocks, trailing commas, unquoted keys, and other common LLM JSON mistakes.

Parameters:

text (str) – Response text that may contain JSON.

Returns:

Parsed JSON object or None if no valid JSON found.

Return type:

Any | None

extract_json_field(data, path)

Extract a value from JSON using dot notation path.

Supports: - Simple fields: “field_name” - Nested objects: “object.field” - Array indices: “array.0” - Combined: “object.array.0.field”

Parameters:

• data (Any) – Parsed JSON data (dict or list)

• path (str) – Dot-separated path to extract

Returns:

Extracted value as string, or empty string if not found

Return type:

str

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]

interpolate_prompt(prompt, history, strict=False)

Replace {{prompt_name.response}} patterns with actual content.

Parameters:

• prompt (str) – Template containing {{}} patterns

• history (dict[str, str]) – Dict mapping prompt_name to response text

• strict (bool) – If True, raise ValueError on unknown references instead of silently replacing with empty string.

Returns:

Tuple of (resolved_prompt, set_of_interpolated_prompt_names)

Return type:

tuple[str, set[str]]

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

resolve_graph_prompt(prompt_spec, results_by_name)

Resolve interpolation and history injection for a graph node.

Performs two-phase prompt assembly:

1. Interpolation: Resolves {{name.response}} patterns using results from earlier DAG levels.

2. History injection: For each name in prompt_spec["history"], formats the prior Q&A as <conversation_history> XML. Entries already interpolated are deduplicated.

Parameters:

• prompt_spec (dict[str, Any]) – The prompt dict with keys prompt, history, etc.

• results_by_name (dict[str, dict[str, Any]]) – Accumulated results from prior DAG levels.

Returns:

Tuple of (resolved_prompt, set_of_interpolated_names).

Return type:

tuple[str, set[str]]

run_sync(coro)

Run an async coroutine from synchronous code.

If no event loop is running, delegates to asyncio.run(). If an event loop is already running (e.g. inside a Jupyter notebook or an existing async framework), runs the coroutine in a background thread to avoid RuntimeError.

Parameters:

coro (Coroutine[Any, Any, T]) – The coroutine to execute.

Returns:

The coroutine’s result.

Return type:

T

should_skip_for_failed_deps(node, results_by_name, nodes)

Check if a node should be skipped because any dependency failed.

Only failed dependencies trigger a skip. Skipped dependencies do NOT cascade — a node whose only dependency was skipped will still execute (its condition will determine whether it should proceed).

Parameters:

• node (PromptNode) – The prompt node to check.

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

• nodes (dict[int, PromptNode]) – All graph nodes indexed by sequence number.

Returns:

Tuple of (should_skip, reason_string).

Return type:

tuple[bool, str]