pyrit.analytics

Analytics module for PyRIT conversation and result analysis.

Functions¶

analyze_results¶

analyze_results(attack_results: list[AttackResult], include_outcome_statistics: bool = False) → _AnalysisResult[AttackStats] | _AnalysisResult[OutcomeStatistics]

Analyze a list of AttackResult objects and return overall and grouped statistics.

This API remains supported. Both output shapes use the shared outcome calculator and count the supplied results without changing their grouping or selecting retries.

Returns:

• _AnalysisResult[AttackStats] | _AnalysisResult[OutcomeStatistics] — Overall statistics under “Overall” and per-attack-type statistics under “By_attack_identifier”. The opt-in changes values, not dictionary keys.

Raises:

• ValueError — if attack_results is empty.

• TypeError — if any element is not an AttackResult.

Examples:

• Attack-result analytics

combine_outcome_statistics¶

combine_outcome_statistics(statistics: Iterable[AttackStats]) → OutcomeStatistics

Sum disjoint populations’ outcome counts and recalculate both rates.

Never average subgroup percentages. The caller must ensure its populations are disjoint; overlapping harm/converter groups cannot reconstruct a cohort. Existing AttackStats objects are accepted without changing their legacy shape.

Returns:

• OutcomeStatistics — Statistics from the combined counts, including an empty population.

Raises:

• ValueError — If any input count is invalid, even if another input would offset it.

Examples:

• Attack-result analytics

compute_outcome_statistics¶

compute_outcome_statistics(counts: Mapping[str, int] | Mapping[AttackOutcome, int]) → OutcomeStatistics

Calculate both success rates for an already selected population.

This function neither queries memory nor chooses result IDs, retries, roles, or scenario units. success_rate_decided (also success_rate) uses success + failure; success_rate_all uses every outcome. Both are proportions, not rounded percentages.

Returns:

• OutcomeStatistics — Counts, both denominator policies, and whole-population shares. Empty denominators produce None; empty outcome shares are zero.

Raises:

• ValueError — If an outcome is unsupported or a count is not a nonnegative integer.

Examples:

• Attack-result analytics

compute_producer_counts¶

compute_producer_counts(attempts: Iterable[_CountableAttempt]) → ScenarioProducerCounts

Compute role-aware attempt, error, and retry accounting across attempts.

Returns:

• ScenarioProducerCounts — Role-specific attempts, errors, and retries.

compute_scenario_statistics¶

compute_scenario_statistics(scenario_result: ScenarioResult, plan: ScenarioRunPlan | None = None, use_saved_plan: bool = True) → ScenarioExecutionStatistics

Calculate effective execution-unit statistics for a scenario result.

With a run plan, counts cover the planned units, and attempts that match no planned unit are reported as unattributed. Without one (legacy results), every resolved unit counts and display groups come from display_group_map. Producer counts include all saved attempts by recorded atomic attack name, even when a configuration matches no planned unit. Display groups use a resolved planned group when available, otherwise the saved display-group mapping.

Returns:

• ScenarioExecutionStatistics — Overall, per atomic attack, and per display group counts.

Examples:

• Attack-result analytics

compute_technique_stats_async¶

compute_technique_stats_async(technique_eval_hashes: Sequence[str], scenario_result_id: str | None = None, targeted_harm_categories: Sequence[str] | None = None, memory: MemoryInterface | None = None, include_outcome_statistics: bool = False) → dict[str, AttackStats] | dict[str, OutcomeStatistics]

Compute per-technique outcome statistics from persisted attack results.

Queries memory for all AttackResult rows whose atomic_attack_identifier.eval_hash matches one of technique_eval_hashes, then aggregates outcomes into per-technique AttackStats. The eval hash is auto-stamped on every persisted result by AtomicAttackEvaluationIdentifier and is the canonical primitive for behavioral-equivalence aggregation (seeds excluded, scorer excluded, only behavior-relevant target params included).

This async API remains supported. The output opt-in does not change its memory query, cohort selection, or behavior used by adaptive technique selectors.

Returns:

• dict[str, AttackStats] | dict[str, OutcomeStatistics] — dict[str, AttackStats] | dict[str, OutcomeStatistics]: Stats per technique eval hash, including both rates when requested. Hashes with no results are omitted.

Examples:

• Attack-result analytics

count_execution_units¶

count_execution_units(units: Iterable[ScenarioExecutionUnit], attempts_by_unit: Mapping[ScenarioExecutionUnit, Sequence[_CountableAttempt]], planned: int | None, producer_attempts: Iterable[_CountableAttempt] | None = None) → ScenarioProgressCounts

Count effective execution units from chronologically ordered attempts.

Attempts are ordered oldest first. The latest top-level result decides the unit’s outcome; children cannot complete their compound unit. Historical pressure excludes orchestration envelopes when execution attempts exist for the unit.

Returns:

• ScenarioProgressCounts — Shared statistics for latest unit outcomes plus historical errors and retries.

get_cached_results_for_technique¶

get_cached_results_for_technique(memory_interface: MemoryInterface, technique_eval_hash: str, objective_target_eval_hash: str, additional_filters: Sequence[IdentifierFilter] | None = None) → list[AttackResult]

Return cached AttackResults matching a (technique × objective target) pair.

Memory is queried for AttackResults whose stamped atomic_attack_identifier.eval_hash equals technique_eval_hash, then results are filtered in Python to those whose nested objective target produces the requested objective_target_eval_hash (computed via ObjectiveTargetEvaluationIdentifier). Returned results are sorted newest-first by timestamp so the most recent is at index 0.

No scenario scoping is applied; this is a behavioral cache spanning every run that produced the same (technique × target) combination. Callers that need scenario-level scoping should pass additional IdentifierFilters or filter the returned list themselves.

Returns:

• list[AttackResult] — list[AttackResult]: Matching attack results sorted newest-first. Empty list if no cache hit.

get_cached_results_for_technique_async¶

get_cached_results_for_technique_async(memory_interface: MemoryInterface, technique_eval_hash: str, objective_target_eval_hash: str, additional_filters: Sequence[IdentifierFilter] | None = None) → list[AttackResult]

Return cached AttackResults matching a (technique × objective target) pair.

Memory is queried for AttackResults whose stamped atomic_attack_identifier.eval_hash equals technique_eval_hash, then results are filtered in Python to those whose nested objective target produces the requested objective_target_eval_hash (computed via ObjectiveTargetEvaluationIdentifier). Returned results are sorted newest-first by timestamp so the most recent is at index 0.

No scenario scoping is applied; this is a behavioral cache spanning every run that produced the same (technique × target) combination. Callers that need scenario-level scoping should pass additional IdentifierFilters or filter the returned list themselves.

Returns:

• list[AttackResult] — list[AttackResult]: Matching attack results sorted newest-first. Empty list if no cache hit.

get_lazy_dir¶

get_lazy_dir(module_globals: dict[str, Any], exports: Mapping[str, LazyExport]) → list[str]

Return package attributes and unresolved public exports.

Returns:

• list[str] — list[str]: Sorted package attribute names.

resolve_lazy_export¶

resolve_lazy_export(name: str, module_name: str, module_globals: dict[str, Any], exports: Mapping[str, LazyExport]) → Any

Resolve and cache one package export.

Returns:

• Any — The resolved public value.

Raises:

• AttributeError — If name is not a declared lazy export.

ApproximateTextMatching¶

Bases: TextMatching

Approximate text matching using n-gram overlap.

This strategy computes the proportion of character n-grams from the target that are present in the text. Useful for detecting partial matches, encoded content, or text with variations.

Constructor Parameters:

Methods:

get_identifier_params¶

get_identifier_params() → dict[str, Any]

Return the configuration that determines matching behavior.

Returns:

• dict[str, Any] — dict[str, Any]: Behavioral parameters for scorer identifiers.

get_overlap_score¶

get_overlap_score(target: str, text: str) → float

Get the n-gram overlap score without threshold comparison.

Useful for getting detailed scoring information.

Returns:

• float — The n-gram overlap score between target and text.

is_match¶

is_match(target: str, text: str) → bool

Check if target approximately matches text using n-gram overlap.

Returns:

• bool — True if n-gram overlap score exceeds threshold, False otherwise.

AttackResultAnalytics¶

Analyze every distinct saved result ID, not conversations or scenario execution units.

Facades sharing a memory instance share one loop-bound execution budget. Reuse a long-lived SDK owner, then await close_async on that same loop before disposing or replacing memory. Closing drains and terminates that shared controller, including work submitted through other facades. No authorization, in-flight reuse, or persistent result caching is performed.

Constructor Parameters:

Examples:

• Attack-result analytics

Methods:

close_async¶

close_async() → None

Drain the backend’s shared controller; never dispose caller-owned memory.

Close on the owning loop, after all users of this memory’s analytics lifetime have finished. Cancellation of this await is delayed until actual operation/session cleanup completes. New controllers remain forbidden during draining. Used facades cannot reopen; construct a new SDK instance after closing to begin another lifetime. Closing an unused facade is a no-op for other facades and still permanently closes that unused facade.

facets_async¶

facets_async(query: AttackAnalyticsFacetQuery) → AttackAnalyticsFacets

Look up one facet page without calculating a report or fetching other facets.

Returns:

• AttackAnalyticsFacets — Typed options and the next offset, if any.

query_async¶

query_async(query: AttackAnalyticsQuery | None = None) → AttackAnalyticsReport

Compute a coherent report and its first lightweight result page.

Returns:

• AttackAnalyticsReport — Caller-owned counts, rates, annotations and exact additional drill-down predicates. No snapshot is retained for later calls.

Raises:

• AnalyticsBusyException — If admission is full/expired or this lifetime is closing.

• AnalyticsTimeoutException — If the shared database/SDK execution budget expires.

• AnalyticsDataException — If stored data cannot be represented faithfully.

• RuntimeError — If used from a different event loop.

results_async¶

results_async(query: AttackAnalyticsResultsQuery | None = None) → AttackAnalyticsResults

Fetch a fresh metadata page without recalculating any report.

Returns:

• AttackAnalyticsResults — Caller-owned result projections with their own freshness timestamp.

Raises:

• ValueError — If the cursor is malformed or belongs to different filters.

• AnalyticsBusyException — If quick-query admission fails or shutdown has begun.

• AnalyticsTimeoutException — If the operation exceeds its execution budget.

AttackStats¶

Compatible six-field statistics for attack analysis results.

ConversationAnalytics¶

Handles analytics operations on conversation data, such as finding similar chat messages based on conversation history or embedding similarity.

Constructor Parameters:

Methods:

get_prompt_entries_with_same_converted_content¶

get_prompt_entries_with_same_converted_content(chat_message_content: str) → list[ConversationMessageWithSimilarity]

Retrieve chat messages that have the same converted content.

Returns:

• list[ConversationMessageWithSimilarity] — list[ConversationMessageWithSimilarity]: A list of ConversationMessageWithSimilarity objects representing

• list[ConversationMessageWithSimilarity] — the similar chat messages based on content.

get_prompt_entries_with_same_converted_content_async¶

get_prompt_entries_with_same_converted_content_async(chat_message_content: str) → list[ConversationMessageWithSimilarity]

Retrieve chat messages that have the same converted content.

Returns:

• list[ConversationMessageWithSimilarity] — list[ConversationMessageWithSimilarity]: A list of ConversationMessageWithSimilarity objects representing

• list[ConversationMessageWithSimilarity] — the similar chat messages based on content.

get_similar_chat_messages_by_embedding¶

get_similar_chat_messages_by_embedding(chat_message_embedding: list[float], threshold: float = 0.8) → list[EmbeddingMessageWithSimilarity]

Retrieve chat messages that are similar to the given embedding based on cosine similarity.

Returns:

• list[EmbeddingMessageWithSimilarity] — list[ConversationMessageWithSimilarity]: A list of ConversationMessageWithSimilarity objects representing

• list[EmbeddingMessageWithSimilarity] — the similar chat messages based on embedding similarity.

get_similar_chat_messages_by_embedding_async¶

get_similar_chat_messages_by_embedding_async(chat_message_embedding: list[float], threshold: float = 0.8) → list[EmbeddingMessageWithSimilarity]

Retrieve chat messages that are similar to the given embedding based on cosine similarity.

Returns:

• list[EmbeddingMessageWithSimilarity] — list[ConversationMessageWithSimilarity]: A list of ConversationMessageWithSimilarity objects representing

• list[EmbeddingMessageWithSimilarity] — the similar chat messages based on embedding similarity.

ExactTextMatching¶

Bases: TextMatching

Exact substring matching strategy.

Checks if the target string is present in the text as a substring.

Constructor Parameters:

Methods:

get_identifier_params¶

get_identifier_params() → dict[str, Any]

Return the configuration that determines matching behavior.

Returns:

• dict[str, Any] — dict[str, Any]: Behavioral parameters for scorer identifiers.

is_match¶

is_match(target: str, text: str) → bool

Check if target string is present in text.

Returns:

• bool — True if target is found in text, False otherwise.

TextMatching¶

Bases: Protocol

Protocol for text matching strategies.

Classes implementing this protocol must provide an is_match method that checks if a target string matches text according to some strategy.

Matchers may additionally expose get_identifier_params() with stable, JSON-serializable behavioral parameters for use in scorer identifiers.

Methods:

is_match¶

is_match(target: str, text: str) → bool

Check if target matches text according to the strategy.

Returns:

• bool — True if target matches text according to the strategy, False otherwise.