tools.feature_atlas.run_interaction_analysis_swarm module

Step 5: Gemini Flash swarm for interaction analysis.

Loads interaction prompts (already sorted by code evidence count), selects the top N (default 50), and dispatches parallel Gemini Flash agents to analyze each interaction. Handles JSON parse retries with backoff and validates/clamps the returned scores.

Outputs outputs/interaction_analyses.jsonl.

Usage:

python -m tools.feature_atlas.run_interaction_analysis_swarm

# skull fire infinity – THE SWARM READS THE BODY

async tools.feature_atlas.run_interaction_analysis_swarm.run_analysis(config=None)

Select the top feature pairs and run the Gemini Flash analysis swarm.

Orchestrates the whole step: it loads the prompt records, picks which pairs to analyze, fans out one Gemini agent per pair under a concurrency cap, and collects the successful results. Pair selection prefers candidates whose code evidence count meets the configured min_evidence_threshold; if fewer than top_n_pairs qualify, it pads up to top_n_pairs with the highest-evidence remaining pairs so the swarm always runs at full requested width when prompts are available.

Loads the config via _load_config() when none is passed, reads prompts via _load_prompts(), builds an asyncio.Semaphore from the swarm.max_concurrency setting, and dispatches _analyze_pair() for each selected pair with asyncio.gather() (return_exceptions=True). Each pair’s LLM call goes through the local proxy used by tools.feature_atlas.extract_features_swarm._gemini_generate(); dict results are kept and exceptions are logged and counted as failures via the module logger. No Redis, FalkorDB, KG, event bus, or output-file writes happen here – persistence is left to async_main().

Called by async_main() as the core of the command-line run.

Parameters:

config (dict[str, Any] | None) – The run configuration; loaded from config.yaml via _load_config() when None. Its analysis block supplies top_n_pairs and min_evidence_threshold and its swarm block supplies max_concurrency.

Returns:

The successfully completed analysis records (failed pairs are dropped, not represented).

Return type:

list[dict[str, Any]]

async tools.feature_atlas.run_interaction_analysis_swarm.async_main()

Run the analysis swarm end to end, persist results, and print a summary.

The asynchronous entry point for this step: it loads the configuration, runs the swarm, writes the completed analyses to disk, and prints a human-readable summary (counts, average and maximum risk and synergy scores, and the top three risk and synergy pairs) plus the elapsed time. This is what turns the in-memory analysis list into the durable outputs/interaction_analyses.jsonl artifact that later Atlas steps import.

Loads config via _load_config(), delegates the actual analysis to run_analysis() (which drives the Gemini Flash agents through the local proxy), then creates the output directory and writes each record as one JSON line to the module-level _OUTPUT_PATH. The only side effects are that filesystem write and stdout printing; no Redis, FalkorDB, KG, or event bus is touched here.

Called by main() via asyncio.run(); main in turn runs under this module’s __main__ guard.

Return type:

None

Returns:

None

tools.feature_atlas.run_interaction_analysis_swarm.main()

Configure logging and run the analysis swarm from a synchronous context.

The synchronous command-line entry point invoked when this module is run as python -m tools.feature_atlas.run_interaction_analysis_swarm. It installs a basic INFO-level logging configuration (timestamped, level-tagged lines) so the swarm’s progress and warnings are visible, then drives the async pipeline by handing async_main() to asyncio.run().

Side effects are limited to the global logging configuration and whatever async_main() does (the JSONL write and stdout summary); no Redis, FalkorDB, KG, or event bus is touched here.

Called only from this module’s if __name__ == "__main__": guard.

Return type:

None

Returns:

None