core.history

History tracking strategies for FFAI interactions.

History tracking strategies for FFAI interactions.

class ConversationHistory[source]

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)[source]

Append an assistant turn to the history.

Parameters:

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

Return type:

None

add_turn_user(content)[source]

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()[source]

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 HistoryRecorder(context, permanent_history, ordered_history, memory=None, memory_persist=False, memory_persist_path=None)[source]

Bases: object

Records interactions to all 5 history stores in a single operation.

Owns the raw history and clean_history lists and coordinates writes to PermanentHistory, OrderedPromptHistory, and ResponseContext. Optionally embeds each recorded Q+A pair into a Memory instance on a fire-and-forget background thread.

Callers invoke record() instead of manually writing to 5 separate stores.

Parameters:

  • context (ResponseContext) – The ResponseContext for prompt_attr_history recording.

  • permanent_history (PermanentHistory) – The PermanentHistory for chronological turns.

  • ordered_history (OrderedPromptHistory) – The OrderedPromptHistory for named interactions.

  • memory (Memory | None) – Optional Memory instance. When provided, each successful record() call submits an embedding task to a dedicated single-worker thread pool. Failed embeds are logged at WARNING and dropped; they never propagate to the caller.

  • memory_persist (bool) – If True, persist the memory store to Parquet after each successful embed. Requires memory_persist_path to be set.

  • memory_persist_path (str | None) – Fully-qualified file path for the Parquet persistence file. Ignored when memory_persist is False or memory is None. Typically computed by FFAI.__init__() as f"{config.memory.persist_dir}/{config.memory.collection_name}.parquet".

record(prompt, response, model, prompt_name=None, history=None, status='success', resolved_prompt=None, usage=None, metadata=None)[source]

Record an interaction to all 5 history stores.

When a Memory instance is configured and status == "success", the Q+A pair is embedded on a fire-and-forget background thread. Metadata is derived from prompt_name when not explicitly passed.

Parameters:

  • prompt (str) – The resolved prompt text.

  • response (Any) – The cleaned response.

  • model (str) – Model identifier used.

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

  • history (list[str] | None) – List of prompt names this call depends on.

  • status (str) – Execution status (“success”, “skipped”, “failed”). Only "success" triggers embedding.

  • resolved_prompt (str | None) – The fully interpolated prompt sent to the model.

  • usage (Any) – Token usage from the API call.

  • metadata (dict[str, Any] | None) – Optional caller metadata. When None, derived from prompt_name as {"prompt_name": prompt_name} (or {} if prompt_name is also None).

Return type:

None

shutdown()[source]

Shut down the background embed thread pool.

Safe to call multiple times. Called by FFAI.close() on teardown. Pending submitted tasks are not awaited (wait=False); they may complete or be abandoned at interpreter exit.

Return type:

None

class OrderedPromptHistory[source]

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.

__init__()[source]

Initialize an empty OrderedPromptHistory.

Return type:

None

get_effective_prompt_name(prompt_name)[source]

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)[source]

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)[source]

Get all interactions for a specific prompt name.

Parameters:

prompt_name (str)

Return type:

list[Interaction]

get_latest_interaction_by_prompt_name(prompt_name)[source]

Get the most recent interaction for a specific prompt name.

Parameters:

prompt_name (str)

Return type:

Interaction | None

get_all_prompt_names()[source]

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

Return type:

list[str]

get_all_interactions()[source]

Get all interactions in sequence order.

Return type:

list[Interaction]

get_prompt_name_usage_stats()[source]

Get statistics on prompt name usage.

Return type:

dict[str, int]

get_interactions_by_model_and_prompt_name(model, prompt_name)[source]

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

Parameters:

  • model (str)

  • prompt_name (str)

Return type:

list[Interaction]

merge_histories(other)[source]

Merge another OrderedPromptHistory into this one.

Parameters:

other (OrderedPromptHistory) – Another OrderedPromptHistory instance to merge

Return type:

None

to_dict()[source]

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

Return type:

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

get_interaction_by_prompt(prompt)[source]

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)[source]

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)[source]

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[source]

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)[source]

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)[source]

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()[source]

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)[source]

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]]

Classes

class ConversationHistory[source]

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)[source]

Append an assistant turn to the history.

Parameters:

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

Return type:

None

add_turn_user(content)[source]

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()[source]

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 HistoryRecorder(context, permanent_history, ordered_history, memory=None, memory_persist=False, memory_persist_path=None)[source]

Bases: object

Records interactions to all 5 history stores in a single operation.

Owns the raw history and clean_history lists and coordinates writes to PermanentHistory, OrderedPromptHistory, and ResponseContext. Optionally embeds each recorded Q+A pair into a Memory instance on a fire-and-forget background thread.

Callers invoke record() instead of manually writing to 5 separate stores.

Parameters:

  • context (ResponseContext) – The ResponseContext for prompt_attr_history recording.

  • permanent_history (PermanentHistory) – The PermanentHistory for chronological turns.

  • ordered_history (OrderedPromptHistory) – The OrderedPromptHistory for named interactions.

  • memory (Memory | None) – Optional Memory instance. When provided, each successful record() call submits an embedding task to a dedicated single-worker thread pool. Failed embeds are logged at WARNING and dropped; they never propagate to the caller.

  • memory_persist (bool) – If True, persist the memory store to Parquet after each successful embed. Requires memory_persist_path to be set.

  • memory_persist_path (str | None) – Fully-qualified file path for the Parquet persistence file. Ignored when memory_persist is False or memory is None. Typically computed by FFAI.__init__() as f"{config.memory.persist_dir}/{config.memory.collection_name}.parquet".

record(prompt, response, model, prompt_name=None, history=None, status='success', resolved_prompt=None, usage=None, metadata=None)[source]

Record an interaction to all 5 history stores.

When a Memory instance is configured and status == "success", the Q+A pair is embedded on a fire-and-forget background thread. Metadata is derived from prompt_name when not explicitly passed.

Parameters:

  • prompt (str) – The resolved prompt text.

  • response (Any) – The cleaned response.

  • model (str) – Model identifier used.

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

  • history (list[str] | None) – List of prompt names this call depends on.

  • status (str) – Execution status (“success”, “skipped”, “failed”). Only "success" triggers embedding.

  • resolved_prompt (str | None) – The fully interpolated prompt sent to the model.

  • usage (Any) – Token usage from the API call.

  • metadata (dict[str, Any] | None) – Optional caller metadata. When None, derived from prompt_name as {"prompt_name": prompt_name} (or {} if prompt_name is also None).

Return type:

None

shutdown()[source]

Shut down the background embed thread pool.

Safe to call multiple times. Called by FFAI.close() on teardown. Pending submitted tasks are not awaited (wait=False); they may complete or be abandoned at interpreter exit.

Return type:

None

class OrderedPromptHistory[source]

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.

__init__()[source]

Initialize an empty OrderedPromptHistory.

Return type:

None

get_effective_prompt_name(prompt_name)[source]

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)[source]

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)[source]

Get all interactions for a specific prompt name.

Parameters:

prompt_name (str)

Return type:

list[Interaction]

get_latest_interaction_by_prompt_name(prompt_name)[source]

Get the most recent interaction for a specific prompt name.

Parameters:

prompt_name (str)

Return type:

Interaction | None

get_all_prompt_names()[source]

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

Return type:

list[str]

get_all_interactions()[source]

Get all interactions in sequence order.

Return type:

list[Interaction]

get_prompt_name_usage_stats()[source]

Get statistics on prompt name usage.

Return type:

dict[str, int]

get_interactions_by_model_and_prompt_name(model, prompt_name)[source]

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

Parameters:

  • model (str)

  • prompt_name (str)

Return type:

list[Interaction]

merge_histories(other)[source]

Merge another OrderedPromptHistory into this one.

Parameters:

other (OrderedPromptHistory) – Another OrderedPromptHistory instance to merge

Return type:

None

to_dict()[source]

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

Return type:

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

get_interaction_by_prompt(prompt)[source]

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)[source]

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)[source]

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[source]

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)[source]

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)[source]

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()[source]

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)[source]

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]]