From 29352593c2c6b08c043a570fb7793aff10a866d7 Mon Sep 17 00:00:00 2001 From: Varun Nuthalapati Date: Sun, 28 Jun 2026 12:42:25 -0700 Subject: [PATCH] =?UTF-8?q?feat(python):=20add=20AgentOverlapAnalyzer=20?= =?UTF-8?q?=E2=80=94=20Python=20port=20of=20TypeScript=20analyzer?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adds python/src/agent_squad/agent_overlap_analyzer.py as a pure-stdlib Python implementation of the TypeScript AgentOverlapAnalyzer. Uses the same TF-IDF / cosine-similarity approach and identical conflict thresholds (>0.3 → High, >0.1 → Medium, else Low). Includes 12 unit tests covering pairwise overlap, uniqueness scores, conflict levels, edge cases (0 or 1 agent), and identical / orthogonal descriptions. Closes #119 --- .../src/agent_squad/agent_overlap_analyzer.py | 147 ++++++++++++++++++ .../src/tests/test_agent_overlap_analyzer.py | 145 +++++++++++++++++ 2 files changed, 292 insertions(+) create mode 100644 python/src/agent_squad/agent_overlap_analyzer.py create mode 100644 python/src/tests/test_agent_overlap_analyzer.py diff --git a/python/src/agent_squad/agent_overlap_analyzer.py b/python/src/agent_squad/agent_overlap_analyzer.py new file mode 100644 index 00000000..07e55ccb --- /dev/null +++ b/python/src/agent_squad/agent_overlap_analyzer.py @@ -0,0 +1,147 @@ +from __future__ import annotations +import math +import re +from dataclasses import dataclass +from typing import Optional + +_STOPWORDS = frozenset([ + 'a', 'an', 'the', 'and', 'or', 'but', 'in', 'on', 'at', 'to', 'for', + 'of', 'with', 'by', 'is', 'are', 'was', 'were', 'be', 'been', 'being', + 'have', 'has', 'had', 'do', 'does', 'did', 'will', 'would', 'could', + 'should', 'may', 'might', 'shall', 'can', 'this', 'that', 'these', + 'those', 'i', 'you', 'he', 'she', 'it', 'we', 'they', 'what', 'which', + 'who', 'when', 'where', 'how', 'why', 'not', 'from', 'as', 'into', + 'through', 'during', 'before', 'after', 'above', 'below', 'up', 'down', + 'out', 'off', 'over', 'under', 'then', 'once', 'so', +]) + + +@dataclass +class OverlapResult: + overlap_percentage: str + potential_conflict: str # "High" | "Medium" | "Low" + + +@dataclass +class UniquenessScore: + agent: str + uniqueness_score: str + + +@dataclass +class AnalysisResult: + pairwise_overlap: dict[str, OverlapResult] + uniqueness_scores: list[UniquenessScore] + + +class AgentOverlapAnalyzer: + """Analyses description overlap between agents using TF-IDF cosine similarity. + + Mirrors the TypeScript ``AgentOverlapAnalyzer`` (``typescript/src/agentOverlapAnalyzer.ts``). + + Args: + agents: Mapping of agent key → ``{"name": ..., "description": ...}``. + """ + + def __init__(self, agents: dict[str, dict[str, str]]) -> None: + self._agents = agents + + def analyze_overlap(self) -> Optional[AnalysisResult]: + """Run the overlap analysis and print results to stdout. + + Returns: + :class:`AnalysisResult` when two or more agents are present, + ``None`` otherwise. + """ + agent_names = list(self._agents.keys()) + agent_descriptions = [self._agents[k]['description'] for k in agent_names] + + if len(agent_names) < 2: + print("Agent Overlap Analysis requires at least two agents.") + print(f"Current number of agents: {len(agent_names)}") + if len(agent_names) == 1: + print("\nSingle Agent Information:") + print(f"Agent Name: {agent_names[0]}") + print(f"Description: {agent_descriptions[0]}") + return None + + tokenized = [self._tokenize(d) for d in agent_descriptions] + tfidf_vectors = self._build_tfidf(tokenized) + + pairwise_overlap: dict[str, OverlapResult] = {} + for i in range(len(agent_names)): + for j in range(i + 1, len(agent_names)): + similarity = self._cosine_similarity(tfidf_vectors[i], tfidf_vectors[j]) + key = f"{agent_names[i]}__{agent_names[j]}" + pairwise_overlap[key] = OverlapResult( + overlap_percentage=f"{similarity * 100:.2f}%", + potential_conflict="High" if similarity > 0.3 else "Medium" if similarity > 0.1 else "Low", + ) + + uniqueness_scores: list[UniquenessScore] = [] + for i, name in enumerate(agent_names): + similarities: list[float] = [] + for j in range(len(agent_names)): + if i == j: + continue + lo, hi = min(i, j), max(i, j) + key = f"{agent_names[lo]}__{agent_names[hi]}" + result = pairwise_overlap.get(key) + if result: + similarities.append(float(result.overlap_percentage.rstrip('%')) / 100) + avg_sim = sum(similarities) / len(similarities) if similarities else 0.0 + uniqueness_scores.append(UniquenessScore( + agent=name, + uniqueness_score=f"{(1 - avg_sim) * 100:.2f}%", + )) + + print("Pairwise Overlap Results:") + print("_________________________\n") + for key, result in pairwise_overlap.items(): + agent1, agent2 = key.split("__") + print(f"{agent1} - {agent2}:") + print(f"- Overlap Percentage - {result.overlap_percentage}") + print(f"- Potential Conflict - {result.potential_conflict}\n") + + print("\nUniqueness Scores:") + print("_________________\n") + for score in uniqueness_scores: + print(f"Agent: {score.agent}, Uniqueness Score: {score.uniqueness_score}") + + return AnalysisResult( + pairwise_overlap=pairwise_overlap, + uniqueness_scores=uniqueness_scores, + ) + + @staticmethod + def _tokenize(text: str) -> list[str]: + tokens = re.split(r'\W+', text.lower()) + return [t for t in tokens if t and t not in _STOPWORDS] + + @staticmethod + def _build_tfidf(documents: list[list[str]]) -> list[dict[str, float]]: + n = len(documents) + + tf_vectors: list[dict[str, float]] = [] + for doc in documents: + counts: dict[str, int] = {} + for word in doc: + counts[word] = counts.get(word, 0) + 1 + total = len(doc) or 1 + tf_vectors.append({w: c / total for w, c in counts.items()}) + + df: dict[str, int] = {} + for doc in documents: + for word in set(doc): + df[word] = df.get(word, 0) + 1 + idf = {w: math.log((n + 1) / (cnt + 1)) + 1 for w, cnt in df.items()} + + return [{w: score * idf.get(w, 1.0) for w, score in tf.items()} for tf in tf_vectors] + + @staticmethod + def _cosine_similarity(vec1: dict[str, float], vec2: dict[str, float]) -> float: + terms = set(vec1) | set(vec2) + dot = sum(vec1.get(t, 0.0) * vec2.get(t, 0.0) for t in terms) + mag1 = math.sqrt(sum(v * v for v in vec1.values())) + mag2 = math.sqrt(sum(v * v for v in vec2.values())) + return dot / (mag1 * mag2) if mag1 and mag2 else 0.0 diff --git a/python/src/tests/test_agent_overlap_analyzer.py b/python/src/tests/test_agent_overlap_analyzer.py new file mode 100644 index 00000000..82856d91 --- /dev/null +++ b/python/src/tests/test_agent_overlap_analyzer.py @@ -0,0 +1,145 @@ +import pytest +from agent_squad.agent_overlap_analyzer import ( + AgentOverlapAnalyzer, + AnalysisResult, + OverlapResult, + UniquenessScore, +) + +TRAVEL_AGENTS = { + "flight_agent": { + "name": "Flight Agent", + "description": "Helps users search, book, and manage airline flights and tickets", + }, + "hotel_agent": { + "name": "Hotel Agent", + "description": "Assists with hotel reservations, room selection, and accommodation bookings", + }, + "weather_agent": { + "name": "Weather Agent", + "description": "Provides weather forecasts and climate information for travel destinations", + }, +} + +SIMILAR_AGENTS = { + "agent_a": { + "name": "Agent A", + "description": "Helps users book flights and airline tickets for travel", + }, + "agent_b": { + "name": "Agent B", + "description": "Assists users to book flights and airline reservations for travel", + }, +} + + +def test_analyze_overlap_returns_analysis_result(): + analyzer = AgentOverlapAnalyzer(TRAVEL_AGENTS) + result = analyzer.analyze_overlap() + assert isinstance(result, AnalysisResult) + + +def test_pairwise_overlap_has_correct_number_of_pairs(): + analyzer = AgentOverlapAnalyzer(TRAVEL_AGENTS) + result = analyzer.analyze_overlap() + # 3 agents → 3 pairs (C(3,2)) + assert len(result.pairwise_overlap) == 3 + + +def test_pairwise_keys_are_formatted_with_double_underscore(): + analyzer = AgentOverlapAnalyzer(TRAVEL_AGENTS) + result = analyzer.analyze_overlap() + for key in result.pairwise_overlap: + assert "__" in key + parts = key.split("__") + assert len(parts) == 2 + assert parts[0] in TRAVEL_AGENTS + assert parts[1] in TRAVEL_AGENTS + + +def test_overlap_percentage_is_valid_string(): + analyzer = AgentOverlapAnalyzer(TRAVEL_AGENTS) + result = analyzer.analyze_overlap() + for overlap in result.pairwise_overlap.values(): + assert isinstance(overlap, OverlapResult) + assert overlap.overlap_percentage.endswith("%") + value = float(overlap.overlap_percentage.rstrip("%")) + assert 0.0 <= value <= 100.0 + + +def test_potential_conflict_levels_are_valid(): + analyzer = AgentOverlapAnalyzer(TRAVEL_AGENTS) + result = analyzer.analyze_overlap() + valid_levels = {"High", "Medium", "Low"} + for overlap in result.pairwise_overlap.values(): + assert overlap.potential_conflict in valid_levels + + +def test_uniqueness_scores_count_matches_agent_count(): + analyzer = AgentOverlapAnalyzer(TRAVEL_AGENTS) + result = analyzer.analyze_overlap() + assert len(result.uniqueness_scores) == len(TRAVEL_AGENTS) + + +def test_uniqueness_score_is_valid_string(): + analyzer = AgentOverlapAnalyzer(TRAVEL_AGENTS) + result = analyzer.analyze_overlap() + for score in result.uniqueness_scores: + assert isinstance(score, UniquenessScore) + assert score.uniqueness_score.endswith("%") + value = float(score.uniqueness_score.rstrip("%")) + assert 0.0 <= value <= 100.0 + + +def test_high_conflict_for_very_similar_agents(): + analyzer = AgentOverlapAnalyzer(SIMILAR_AGENTS) + result = analyzer.analyze_overlap() + overlap = list(result.pairwise_overlap.values())[0] + assert overlap.potential_conflict == "High" + + +def test_returns_none_for_single_agent(capsys): + agents = {"solo": {"name": "Solo", "description": "A single agent"}} + analyzer = AgentOverlapAnalyzer(agents) + result = analyzer.analyze_overlap() + assert result is None + output = capsys.readouterr().out + assert "at least two agents" in output + assert "solo" in output + + +def test_returns_none_for_empty_agents(capsys): + analyzer = AgentOverlapAnalyzer({}) + result = analyzer.analyze_overlap() + assert result is None + output = capsys.readouterr().out + assert "at least two agents" in output + + +def test_identical_descriptions_produce_high_overlap(): + agents = { + "a1": {"name": "A1", "description": "handles customer billing and payment processing"}, + "a2": {"name": "A2", "description": "handles customer billing and payment processing"}, + } + analyzer = AgentOverlapAnalyzer(agents) + result = analyzer.analyze_overlap() + overlap = list(result.pairwise_overlap.values())[0] + assert overlap.potential_conflict == "High" + assert float(overlap.overlap_percentage.rstrip("%")) > 90.0 + + +def test_completely_different_descriptions_produce_low_overlap(): + agents = { + "billing": { + "name": "Billing", + "description": "processes invoices payments transactions refunds financial accounting ledger", + }, + "weather": { + "name": "Weather", + "description": "forecasts temperature precipitation humidity wind climate meteorology", + }, + } + analyzer = AgentOverlapAnalyzer(agents) + result = analyzer.analyze_overlap() + overlap = list(result.pairwise_overlap.values())[0] + assert overlap.potential_conflict == "Low"