Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
147 changes: 147 additions & 0 deletions python/src/agent_squad/agent_overlap_analyzer.py
Original file line number Diff line number Diff line change
@@ -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
145 changes: 145 additions & 0 deletions python/src/tests/test_agent_overlap_analyzer.py
Original file line number Diff line number Diff line change
@@ -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"