agentconnect.core.registry.capability_discovery module

Capability discovery functionality for the AgentConnect framework.

This module provides functions for searching and discovering agent capabilities, including semantic search using embeddings and simpler string matching methods. It leverages LangChain’s vector stores for efficient semantic search.

custom_showwarning(message, category, filename, lineno, file=None, line=None)

Custom warning handler to suppress relevance score warnings.

Parameters:

  • message – The warning message

  • category – The warning category

check_semantic_search_requirements()

Check if the required packages for semantic search are installed.

Return type:

Dict[str, bool]

Returns:

Dictionary indicating which vector store backends are available

calculate_similarity(text1, text2)

Calculate simple Jaccard similarity between two texts.

Parameters:

  • text1 (str) – First text

  • text2 (str) – Second text

Return type:

float

Returns:

Similarity score between 0 and 1

cosine_similarity(vec1, vec2)

Calculate cosine similarity between two vectors.

Parameters:

  • vec1 – First vector

  • vec2 – Second vector

Returns:

Cosine similarity between the vectors

class CapabilityDiscoveryService(vector_store_config=None)

Bases: object

Service for discovering agent capabilities through various search methods.

This class provides methods for finding agents based on their capabilities, including exact string matching and semantic search using vector stores.

Parameters:

vector_store_config (Dict[str, Any])

async initialize_embeddings_model()

Initialize the embeddings model for semantic search.

This should be called after agents have been registered to precompute embeddings for all existing capabilities.

async update_capability_embeddings_cache(registration)

Update capability embeddings for a registration.

Parameters:

registration (AgentRegistration) – Registration information for the agent

Return type:

None

clear_agent_embeddings_cache(agent_id)

Clear the embeddings cache for a specific agent.

Parameters:

agent_id (str) – ID of the agent to clear cache for

Return type:

None

async precompute_all_capability_embeddings(agent_registrations)

Precompute embeddings for all existing capabilities.

Parameters:

agent_registrations (Dict[str, AgentRegistration]) – Dictionary of agent registrations

Return type:

None

async find_by_capability_name(capability_name, agent_registrations, capabilities_index, limit=10, similarity_threshold=0.1)

Find agents by capability name (simple string matching).

Parameters:

  • capability_name (str) – Name of the capability to search for

  • agent_registrations (Dict[str, AgentRegistration]) – Dictionary of agent registrations

  • capabilities_index (Dict[str, Set[str]]) – Index of agent capabilities

  • limit (int) – Maximum number of results to return (default: 10)

  • similarity_threshold (float) – Minimum similarity score to include in results (default: 0.1)

Return type:

List[AgentRegistration]

Returns:

List of agent registrations with the specified capability

async find_by_capability_semantic(capability_description, agent_registrations, limit=10, similarity_threshold=0.1)

Find agents by capability description using semantic search.

Parameters:

  • capability_description (str) – Description of the capability to search for

  • agent_registrations (Dict[str, AgentRegistration]) – Dictionary of agent registrations

  • limit (int) – Maximum number of results to return (default: 10)

  • similarity_threshold (float) – Minimum similarity score to include in results (default: 0.1) For negative scores (distance metrics), this becomes a maximum threshold (lower absolute value means more similar)

Return type:

List[Tuple[AgentRegistration, float]]

Returns:

List of tuples containing agent registrations and similarity scores

async save_vector_store(path)

Save the vector store to disk for faster loading in the future.

Parameters:

path (str) – Directory path to save the vector store

Return type:

bool

Returns:

True if successful, False otherwise

async load_vector_store(path, embeddings_model=None)

Load the vector store from disk.

Parameters:

  • path (str) – Directory path to load the vector store from

  • embeddings_model – Optional embeddings model to use (if None, uses the current one)

Return type:

bool

Returns:

True if successful, False otherwise