PineconeAsyncio¶
- class pinecone.PineconeAsyncio(api_key: str | None = None, host: str | None = None, proxy_url: str | None = None, ssl_ca_certs: str | None = None, ssl_verify: bool | None = None, additional_headers: Dict[str, str] | None = {}, **kwargs)[source]¶
PineconeAsyncio
is an asyncio client for interacting with Pinecone’s control plane API.This class implements methods for managing and interacting with Pinecone resources such as collections and indexes.
To perform data operations such as inserting and querying vectors, use the
IndexAsyncio
class.import asyncio from pinecone import Pinecone async def main(): pc = Pinecone() async with pc.IndexAsyncio(host="my-index.pinecone.io") as idx: await idx.upsert(vectors=[(1, [1, 2, 3]), (2, [4, 5, 6])]) asyncio.run(main())
- PineconeAsyncio.__init__(api_key: str | None = None, host: str | None = None, proxy_url: str | None = None, ssl_ca_certs: str | None = None, ssl_verify: bool | None = None, additional_headers: Dict[str, str] | None = {}, **kwargs)[source]¶
The
PineconeAsyncio
class is the main entry point for interacting with Pinecone using asyncio. It is used to create, delete, and manage your indexes and collections. Except for needing to useasync with
when instantiating the client andawait
when calling its methods, the functionality provided by this class is extremely similar to the functionality of thePinecone
class.- Parameters:
api_key (str, optional) – The API key to use for authentication. If not passed via kwarg, the API key will be read from the environment variable
PINECONE_API_KEY
.host (str, optional) – The control plane host to connect to.
proxy_url (str, optional) – The URL of the proxy to use for the connection. Default:
None
proxy_headers (Dict[str, str], optional) – Additional headers to pass to the proxy. Use this if your proxy setup requires authentication. Default:
{}
ssl_ca_certs (str, optional) – The path to the SSL CA certificate bundle to use for the connection. This path should point to a file in PEM format. Default:
None
ssl_verify (bool, optional) – SSL verification is performed by default, but can be disabled using the boolean flag. Default:
True
config (pinecone.config.Config, optional) – A
pinecone.config.Config
object. If passed, theapi_key
andhost
parameters will be ignored.additional_headers (Dict[str, str], optional) – Additional headers to pass to the API. Default:
{}
Managing the async context
The
PineconeAsyncio
class relies on an underlyingaiohttp
ClientSession
to make asynchronous HTTP requests. To ensure that the session is properly closed, you should use theasync with
syntax when creating aPineconeAsyncio
object. This will ensure that the session is properly closed when the context is exited.import asyncio from pinecone import PineconeAsyncio async def main(): async with PineconeAsyncio(api_key='YOUR_API_KEY') as pc: # Do async things index_list = await pc.list_indexes() asyncio.run(main())
As an alternative, if you prefer to avoid code with a nested appearance and are willing to manage cleanup yourself, you can await the
close()
method to close the session when you are done.import asyncio from pinecone import PineconeAsyncio async def main(): pc = PineconeAsyncio(api_key='YOUR_API_KEY') # Do async things index_list = await pc.list_indexes() # You're responsible for calling this yourself await pc.close() asyncio.run(main())
Failing to do this may result in error messages appearing from the underlyling aiohttp library.
Configuration with environment variables
If you instantiate the Pinecone client with no arguments, it will attempt to read the API key from the environment variable
PINECONE_API_KEY
.import asyncio from pinecone import PineconeAsyncio async def main(): async with PineconeAsyncio() as pc: # Do async things index_list = await pc.list_indexes() asyncio.run(main())
Configuration with keyword arguments
If you prefer being more explicit in your code, you can also pass the API as
Configuration with environment variables
If you instantiate the Pinecone client with no arguments, it will attempt to read the API key from the environment variable
PINECONE_API_KEY
.import asyncio from pinecone import PineconeAsyncio async def main(): async with PineconeAsyncio() as pc: # Do async things index_list = await pc.list_indexes() asyncio.run(main())
Configuration with environment variables
If you instantiate the Pinecone client with no arguments, it will attempt to read the API key from the environment variable
PINECONE_API_KEY
.import asyncio from pinecone import PineconeAsyncio async def main(): async with PineconeAsyncio() as pc: # Do async things index_list = await pc.list_indexes() asyncio.run(main())
Configuration with keyword arguments
If you prefer being more explicit in your code, you can also pass the API as a keyword argument.
import os import asyncio from pinecone import PineconeAsyncio async def main(): async with Pinecone(api_key=os.environ.get("PINECONE_API_KEY")) as pc: # Do async things index_list = await pc.list_indexes() asyncio.run(main())
Environment variables
The Pinecone client supports the following environment variables:
PINECONE_API_KEY
: The API key to use for authentication. If not passed via
kwarg, the API key will be read from the environment variable
PINECONE_API_KEY
.Proxy configuration
If your network setup requires you to interact with Pinecone via a proxy, you will need to pass additional configuration using optional keyword parameters. These optional parameters are used to configure an SSL context and passed to
aiohttp
, which is the underlying library currently used by the PineconeAsyncio client to make HTTP requests.Here is a basic example:
import asyncio from pinecone import PineconeAsyncio async def main(): async with PineconeAsyncio( api_key='YOUR_API_KEY', proxy_url='https://your-proxy.com' ) as pc: # Do async things index_list = await pc.list_indexes() asyncio.run(main())
Using proxies with self-signed certificates
By default the Pinecone Python client will perform SSL certificate verification using the CA bundle maintained by Mozilla in the certifi package. If your proxy server is using a self-signed certificate, you will need to pass the path to the certificate in PEM format using the
ssl_ca_certs
parameter.import asyncio from pinecone import PineconeAsyncio async def main(): async with PineconeAsyncio( api_key='YOUR_API_KEY', proxy_url='https://your-proxy.com', ssl_ca_certs='path/to/cert-bundle.pem' ) as pc: # Do async things await pc.list_indexes() asyncio.run(main())
Disabling SSL verification
If you would like to disable SSL verification, you can pass the
ssl_verify
parameter with a value ofFalse
. We do not recommend going to production with SSL verification disabled but there are situations where this is useful such as testing with Pinecone Local running in a docker container.import asyncio from pinecone import PineconeAsyncio async def main(): async with PineconeAsyncio( api_key='YOUR_API_KEY', ssl_verify=False ) as pc: if not await pc.has_index('my_index'): await pc.create_index( name='my_index', dimension=1536, metric='cosine', spec=ServerlessSpec(cloud='aws', region='us-west-2') ) asyncio.run(main())
Passing additional headers
If you need to pass additional headers with each request to the Pinecone API, you can do so using the
additional_headers
parameter. This is primarily for internal testing and end-users shouldn’t need to do this unless specifically instructed to do so.import asyncio from pinecone import PineconeAsyncio async def main(): async with PineconeAsyncio( api_key='YOUR_API_KEY', host='https://api-staging.pinecone.io', additional_headers={'X-My-Header': 'my-value'} ) as pc: # Do async things await pc.list_indexes() asyncio.run(main())
DB Control Plane¶
Indexes¶
- async PineconeAsyncio.create_index(name: str, spec: Dict | ServerlessSpec | PodSpec | ByocSpec, dimension: int | None = None, metric: Metric | str | None = 'cosine', timeout: int | None = None, deletion_protection: DeletionProtection | str | None = 'disabled', vector_type: VectorType | str | None = 'dense', tags: Dict[str, str] | None = None) IndexModel [source]¶
Creates a Pinecone index.
- Parameters:
name (str) – The name of the index to create. Must be unique within your project and cannot be changed once created. Allowed characters are lowercase letters, numbers, and hyphens and the name may not begin or end with hyphens. Maximum length is 45 characters.
metric (str, optional) – Type of similarity metric used in the vector index when querying, one of
{"cosine", "dotproduct", "euclidean"}
.spec (Dict) – A dictionary containing configurations describing how the index should be deployed. For serverless indexes, specify region and cloud. For pod indexes, specify replicas, shards, pods, pod_type, metadata_config, and source_collection. Alternatively, use the
ServerlessSpec
orPodSpec
objects to specify these configurations.dimension (int) – If you are creating an index with
vector_type="dense"
(which is the default), you need to specifydimension
to indicate the size of your vectors. This should match the dimension of the embeddings you will be inserting. For example, if you are using OpenAI’s CLIP model, you should usedimension=1536
. Dimension is a required field when creating an index withvector_type="dense"
and should not be passed whenvector_type="sparse"
.timeout (int, optional) – Specify the number of seconds to wait until index gets ready. If None, wait indefinitely; if >=0, time out after this many seconds; if -1, return immediately and do not wait.
deletion_protection (Optional[Literal["enabled", "disabled"]]) – If enabled, the index cannot be deleted. If disabled, the index can be deleted.
vector_type (str, optional) – The type of vectors to be stored in the index. One of
{"dense", "sparse"}
.tags (Optional[Dict[str, str]]) – Tags are key-value pairs you can attach to indexes to better understand, organize, and identify your resources. Some example use cases include tagging indexes with the name of the model that generated the embeddings, the date the index was created, or the purpose of the index.
- Returns:
A
IndexModel
instance containing a description of the index that was created.
Creating a serverless index
import os import asyncio from pinecone import ( PineconeAsyncio, ServerlessSpec, CloudProvider, AwsRegion, Metric, DeletionProtection, VectorType ) async def main(): async with PineconeAsyncio(api_key=os.environ.get("PINECONE_API_KEY")) as pc: await pc.create_index( name="my_index", dimension=1536, metric=Metric.COSINE, spec=ServerlessSpec( cloud=CloudProvider.AWS, region=AwsRegion.US_WEST_2 ), deletion_protection=DeletionProtection.DISABLED, vector_type=VectorType.DENSE, tags={ "model": "clip", "app": "image-search", "env": "testing" } ) asyncio.run(main())
Creating a pod index
import os import asyncio from pinecone import ( Pinecone, PodSpec, PodIndexEnvironment, PodType, Metric, DeletionProtection, VectorType ) async def main(): async with Pinecone(api_key=os.environ.get("PINECONE_API_KEY")) as pc: await pc.create_index( name="my_index", dimension=1536, metric=Metric.COSINE, spec=PodSpec( environment=PodIndexEnvironment.US_EAST4_GCP, pod_type=PodType.P1_X1 ), deletion_protection=DeletionProtection.DISABLED, tags={ "model": "clip", "app": "image-search", "env": "testing" } ) asyncio.run(main())
- async PineconeAsyncio.create_index_for_model(name: str, cloud: CloudProvider | str, region: AwsRegion | GcpRegion | AzureRegion | str, embed: IndexEmbed | CreateIndexForModelEmbedTypedDict, tags: Dict[str, str] | None = None, deletion_protection: DeletionProtection | str | None = 'disabled', timeout: int | None = None) IndexModel [source]¶
- Parameters:
name (str) – The name of the index to create. Must be unique within your project and cannot be changed once created. Allowed characters are lowercase letters, numbers, and hyphens and the name may not begin or end with hyphens. Maximum length is 45 characters.
cloud (str) – The cloud provider to use for the index. One of
{"aws", "gcp", "azure"}
.region (str) – The region to use for the index. Enum objects
AwsRegion
,GcpRegion
, andAzureRegion
are also available to help you quickly set these parameters, but may not be up to date as new regions become available.embed (Union[Dict, IndexEmbed]) – The embedding configuration for the index. This param accepts a dictionary or an instance of the
IndexEmbed
object.tags (Optional[Dict[str, str]]) – Tags are key-value pairs you can attach to indexes to better understand, organize, and identify your resources. Some example use cases include tagging indexes with the name of the model that generated the embeddings, the date the index was created, or the purpose of the index.
deletion_protection (Optional[Literal["enabled", "disabled"]]) – If enabled, the index cannot be deleted. If disabled, the index can be deleted. This setting can be changed with
configure_index
.timeout (Optional[int]) – Specify the number of seconds to wait until index is ready to receive data. If None, wait indefinitely; if >=0, time out after this many seconds; if -1, return immediately and do not wait.
- Returns:
A description of the index that was created.
- Return type:
IndexModel
This method is used to create a Serverless index that is configured for use with Pinecone’s integrated inference models.
The resulting index can be described, listed, configured, and deleted like any other Pinecone index with the
describe_index
,list_indexes
,configure_index
, anddelete_index
methods.After the model is created, you can upsert records into the index with the
upsert_records
method, and search your records with thesearch
method.import asyncio from pinecone import ( PineconeAsyncio, IndexEmbed, CloudProvider, AwsRegion, EmbedModel, Metric, ) async def main(): async with PineconeAsyncio() as pc: if not await pc.has_index("book-search"): desc = await pc.create_index_for_model( name="book-search", cloud=CloudProvider.AWS, region=AwsRegion.US_EAST_1, embed=IndexEmbed( model=EmbedModel.Multilingual_E5_Large, metric=Metric.COSINE, field_map={ "text": "description", }, ) ) asyncio.run(main())
See also:
See the Model Gallery to learn about available models
- PineconeAsyncio.create_index_from_backup(*, name: str, backup_id: str, deletion_protection: DeletionProtection | str | None = 'disabled', tags: Dict[str, str] | None = None, timeout: int | None = None) IndexModel [source]¶
Create an index from a backup.
Call
list_backups
to get a list of backups for your project.- Parameters:
name (str) – The name of the index to create.
backup_id (str) – The ID of the backup to restore.
deletion_protection (Optional[Literal["enabled", "disabled"]]) – If enabled, the index cannot be deleted. If disabled, the index can be deleted. This setting can be changed with
configure_index
.tags (Optional[Dict[str, str]]) – Tags are key-value pairs you can attach to indexes to better understand, organize, and identify your resources. Some example use cases include tagging indexes with the name of the model that generated the embeddings, the date the index was created, or the purpose of the index.
timeout – Specify the number of seconds to wait until index is ready to receive data. If None, wait indefinitely; if >=0, time out after this many seconds; if -1, return immediately and do not wait.
- Returns:
A description of the index that was created.
- Return type:
IndexModel
- async PineconeAsyncio.list_indexes() IndexList [source]¶
- Returns:
Returns an
IndexList
object, which is iterable and contains a list ofIndexModel
objects. TheIndexList
also has a convenience methodnames()
which returns a list of index names for situations where you just want to iterate over all index names.
Lists all indexes in your project.
The results include a description of all indexes in your project, including the index name, dimension, metric, status, and spec.
If you simply want to check whether an index exists, see the
has_index()
convenience method.You can use the
list_indexes()
method to iterate over descriptions of every index in your project.import asyncio from pinecone import PineconeAsyncio async def main(): pc = PineconeAsyncio() available_indexes = await pc.list_indexes() for index in available_indexes: print(index.name) print(index.dimension) print(index.metric) print(index.status) print(index.host) print(index.spec) await pc.close() asyncio.run(main())
- async PineconeAsyncio.describe_index(name: str) IndexModel [source]¶
- Parameters:
name – the name of the index to describe.
- Returns:
Returns an
IndexModel
object which gives access to properties such as the index name, dimension, metric, host url, status, and spec.
Describes a Pinecone index.
Getting your index host url
In a real production situation, you probably want to store the host url in an environment variable so you don’t have to call describe_index and re-fetch it every time you want to use the index. But this example shows how to get the value from the API using describe_index.
import asyncio from pinecone import Pinecone, PineconeAsyncio, Index async def main(): pc = PineconeAsyncio() index_name="my_index" description = await pc.describe_index(name=index_name) print(description) # { # "name": "my_index", # "metric": "cosine", # "host": "my_index-dojoi3u.svc.aped-4627-b74a.pinecone.io", # "spec": { # "serverless": { # "cloud": "aws", # "region": "us-east-1" # } # }, # "status": { # "ready": true, # "state": "Ready" # }, # "vector_type": "dense", # "dimension": 1024, # "deletion_protection": "enabled", # "tags": { # "environment": "production" # } # } print(f"Your index is hosted at {description.host}") await pc.close() async with Pinecone().IndexAsyncio(host=description.host) as idx: await idx.upsert(vectors=[...]) asyncio.run(main())
- async PineconeAsyncio.configure_index(name: str, replicas: int | None = None, pod_type: PodType | str | None = None, deletion_protection: DeletionProtection | str | None = None, tags: Dict[str, str] | None = None)[source]¶
- Param:
name: the name of the Index
- Param:
replicas: the desired number of replicas, lowest value is 0.
- Param:
pod_type: the new pod_type for the index. To learn more about the available pod types, please see Understanding Indexes
- Param:
deletion_protection: If set to ‘enabled’, the index cannot be deleted. If ‘disabled’, the index can be deleted.
- Param:
tags: A dictionary of tags to apply to the index. Tags are key-value pairs that can be used to organize and manage indexes. To remove a tag, set the value to “”. Tags passed to configure_index will be merged with existing tags and any with the value empty string will be removed.
This method is used to modify an index’s configuration. It can be used to:
Scale a pod-based index horizontally using
replicas
Scale a pod-based index vertically using
pod_type
Enable or disable deletion protection using
deletion_protection
Add, change, or remove tags using
tags
Scaling pod-based indexes
To scale your pod-based index, you pass a
replicas
and/orpod_type
param to theconfigure_index
method.pod_type
may be a string or a value from thePodType
enum.import asyncio from pinecone import PineconeAsyncio, PodType async def main(): async with PineconeAsyncio() as pc: await pc.configure_index( name="my_index", replicas=2, pod_type=PodType.P1_X2 ) asyncio.run(main())
After providing these new configurations, you must call
describe_index
to see the status of the index as the changes are applied.Enabling or disabling deletion protection
To enable or disable deletion protection, pass the
deletion_protection
parameter to theconfigure_index
method. When deletion protection is enabled, the index cannot be deleted with thedelete_index
method.import asyncio from pinecone import PineconeAsyncio, DeletionProtection async def main(): async with PineconeAsyncio() as pc: # Enable deletion protection await pc.configure_index( name="my_index", deletion_protection=DeletionProtection.ENABLED ) # Call describe_index to see the change was applied. desc = await pc.describe_index("my_index") assert desc.deletion_protection == "enabled" # Disable deletion protection await pc.configure_index( name="my_index", deletion_protection=DeletionProtection.DISABLED ) asyncio.run(main())
Adding, changing, or removing tags
To add, change, or remove tags, pass the
tags
parameter to theconfigure_index
method. When tags are passed usingconfigure_index
, they are merged with any existing tags already on the index. To remove a tag, set the value of the key to an empty string.import asyncio from pinecone import PineconeAsyncio async def main(): async with PineconeAsyncio() as pc: # Add a tag await pc.configure_index(name="my_index", tags={"environment": "staging"}) # Change a tag await pc.configure_index(name="my_index", tags={"environment": "production"}) # Remove a tag await pc.configure_index(name="my_index", tags={"environment": ""}) # Call describe_index to view the tags are changed await pc.describe_index("my_index") print(desc.tags) asyncio.run(main())
- async PineconeAsyncio.delete_index(name: str, timeout: int | None = None)[source]¶
- Parameters:
name (str) – the name of the index.
timeout (int, optional) – Number of seconds to poll status checking whether the index has been deleted. If None, wait indefinitely; if >=0, time out after this many seconds; if -1, return immediately and do not wait.
Deletes a Pinecone index.
Deleting an index is an irreversible operation. All data in the index will be lost. When you use this command, a request is sent to the Pinecone control plane to delete the index, but the termination is not synchronous because resources take a few moments to be released.
By default the
delete_index
method will block until polling of thedescribe_index
method shows that the delete operation has completed. If you prefer to return immediately and not wait for the index to be deleted, you can passtimeout=-1
to the method.After the delete request is submitted, polling
describe_index
will show that the index transitions into aTerminating
state before eventually resulting in a 404 after it has been removed.This operation can fail if the index is configured with
deletion_protection="enabled"
. In this case, you will need to callconfigure_index
to disable deletion protection before you can delete the index.import asyncio from pinecone import PineconeAsyncio async def main(): pc = PineconeAsyncio() index_name = "my_index" desc = await pc.describe_index(name=index_name) if desc.deletion_protection == "enabled": # If for some reason deletion protection is enabled, you will need to disable it first # before you can delete the index. But use caution as this operation is not reversible # and if somebody enabled deletion protection, they probably had a good reason. await pc.configure_index(name=index_name, deletion_protection="disabled") await pc.delete_index(name=index_name) await pc.close() asyncio.run(main())
- async PineconeAsyncio.has_index(name: str) bool [source]¶
- Parameters:
name – The name of the index to check for existence.
- Returns:
Returns
True
if the index exists,False
otherwise.
Checks if a Pinecone index exists.
import asyncio from pinecone import PineconeAsyncio, ServerlessSpec async def main(): async with PineconeAsyncio() as pc: index_name = "my_index" if not await pc.has_index(index_name): print("Index does not exist, creating...") pc.create_index( name=index_name, dimension=768, metric="cosine", spec=ServerlessSpec(cloud="aws", region="us-west-2") ) asyncio.run(main())
Backups¶
- PineconeAsyncio.create_backup(*, index_name: str, backup_name: str, description: str = '') BackupModel [source]¶
Create a backup of an index.
- Parameters:
index_name (str) – The name of the index to backup.
backup_name (str) – The name to give the backup.
description (str) – Optional description of the backup.
- PineconeAsyncio.list_backups(*, index_name: str | None = None, limit: int | None = 10, pagination_token: str | None = None) BackupList [source]¶
List backups.
If index_name is provided, the backups will be filtered by index. If no index_name is provided, all backups in the projectwill be returned.
- Parameters:
index_name (str) – The name of the index to list backups for.
limit (int) – The maximum number of backups to return.
pagination_token (str) – The pagination token to use for pagination.
Collections¶
- async PineconeAsyncio.create_collection(name: str, source: str)[source]¶
Create a collection from a pod-based index
- Parameters:
name – Name of the collection
source – Name of the source index
- async PineconeAsyncio.list_collections() CollectionList [source]¶
List all collections
import asyncio from pinecone import PineconeAsyncio async def main(): pc = PineconeAsyncio() collections = await pc.list_collections() for collection in collections: print(collection.name) print(collection.source) # You can also iterate specifically over # a list of collection names by calling # the .names() helper. collection_name = "my_collection" collections = await pc.list_collections() if collection_name in collections.names(): print('Collection exists') await pc.close() asyncio.run(main())
- async PineconeAsyncio.describe_collection(name: str)[source]¶
Describes a collection. :param: The name of the collection :return: Description of the collection
import asyncio from pinecone import PineconeAsyncio async def main(): async with PineconeAsyncio() as pc: description = await pc.describe_collection("my_collection") print(description.name) print(description.source) print(description.status) print(description.size) asyncio.run(main())
- async PineconeAsyncio.delete_collection(name: str)[source]¶
Describes a collection. :param: The name of the collection :return: Description of the collection
import asyncio from pinecone import PineconeAsyncio async def main(): async with PineconeAsyncio() as pc: description = await pc.describe_collection("my_collection") print(description.name) print(description.source) print(description.status) print(description.size) asyncio.run(main())
Restore Jobs¶
DB Data Plane¶
- pinecone.db_data.IndexAsyncio¶
alias of
_IndexAsyncio
- IndexAsyncio.__init__(api_key: str, host: str, additional_headers: Dict[str, str] | None = {}, openapi_config=None, **kwargs)¶
- IndexAsyncio.describe_index_stats(filter: Dict[str, str | int | float | bool] | Dict[Literal['$eq'], str | int | float | bool] | Dict[Literal['$ne'], str | int | float | bool] | Dict[Literal['$gt'], int | float] | Dict[Literal['$gte'], int | float] | Dict[Literal['$lt'], int | float] | Dict[Literal['$lte'], int | float] | Dict[Literal['$in'], List[str | int | float | bool]] | Dict[Literal['$nin'], List[str | int | float | bool]] | Dict[Literal['$and'], List[Dict[str, str | int | float | bool] | Dict[Literal['$eq'], str | int | float | bool] | Dict[Literal['$ne'], str | int | float | bool] | Dict[Literal['$gt'], int | float] | Dict[Literal['$gte'], int | float] | Dict[Literal['$lt'], int | float] | Dict[Literal['$lte'], int | float] | Dict[Literal['$in'], List[str | int | float | bool]] | Dict[Literal['$nin'], List[str | int | float | bool]]]] | None = None, **kwargs) IndexDescription ¶
The DescribeIndexStats operation returns statistics about the index’s contents. For example: The vector count per namespace and the number of dimensions.
- Parameters:
filter (Dict[str, Union[str, float, int, bool, List, dict]])
present (If this parameter is)
filter. (the operation only returns statistics for vectors that satisfy the)
<https (See `metadata filtering) –
//www.pinecone.io/docs/metadata-filtering/>_` [optional]
Returns: DescribeIndexStatsResponse object which contains stats about the index.
import asyncio from pinecone import Pinecone, Vector, SparseValues async def main(): pc = Pinecone() async with pc.IndexAsyncio(host="example-dojoi3u.svc.aped-4627-b74a.pinecone.io") as idx: print(await idx.describe_index_stats()) asyncio.run(main())
Vectors¶
- IndexAsyncio.upsert(vectors: List[Vector] | List[Tuple[str, List[float]]] | List[Tuple[str, List[float], Dict[str, str | int | float | List[str] | List[int] | List[float]]]] | List[VectorTypedDict], namespace: str | None = None, batch_size: int | None = None, show_progress: bool = True, **kwargs) UpsertResponse ¶
- Parameters:
vectors (Union[List[Vector], List[VectorTuple], List[VectorTupleWithMetadata], List[VectorTypedDict]]) – A list of vectors to upsert.
namespace (str) – The namespace to write to. If not specified, the default namespace is used. [optional]
batch_size (int) – The number of vectors to upsert in each batch. If not specified, all vectors will be upserted in a single batch. [optional]
show_progress (bool) – Whether to show a progress bar using tqdm. Applied only if batch_size is provided. Default is True.
- Returns:
UpsertResponse, includes the number of vectors upserted.
The upsert operation writes vectors into a namespace. If a new value is upserted for an existing vector id, it will overwrite the previous value.
To upsert in parallel follow this link.
Upserting dense vectors
Note
The dimension of each dense vector must match the dimension of the index.
A vector can be represented in a variety of ways.
import asyncio from pinecone import Pinecone, Vector async def main(): pc = Pinecone() async with pc.IndexAsyncio(host="example-dojoi3u.svc.aped-4627-b74a.pinecone.io") as idx: # A Vector object await idx.upsert( namespace = 'my-namespace', vectors = [ Vector(id='id1', values=[0.1, 0.2, 0.3, 0.4], metadata={'metadata_key': 'metadata_value'}), ] ) # A vector tuple await idx.upsert( namespace = 'my-namespace', vectors = [ ('id1', [0.1, 0.2, 0.3, 0.4]), ] ) # A vector tuple with metadata await idx.upsert( namespace = 'my-namespace', vectors = [ ('id1', [0.1, 0.2, 0.3, 0.4], {'metadata_key': 'metadata_value'}), ] ) # A vector dictionary await idx.upsert( namespace = 'my-namespace', vectors = [ {"id": 1, "values": [0.1, 0.2, 0.3, 0.4], "metadata": {"metadata_key": "metadata_value"}}, ] asyncio.run(main())
Upserting sparse vectors
import asyncio from pinecone import Pinecone, Vector, SparseValues async def main(): pc = Pinecone() async with pc.IndexAsyncio(host="example-dojoi3u.svc.aped-4627-b74a.pinecone.io") as idx: # A Vector object await idx.upsert( namespace = 'my-namespace', vectors = [ Vector(id='id1', sparse_values=SparseValues(indices=[1, 2], values=[0.2, 0.4])), ] ) # A dictionary await idx.upsert( namespace = 'my-namespace', vectors = [ {"id": 1, "sparse_values": {"indices": [1, 2], "values": [0.2, 0.4]}}, ] ) asyncio.run(main())
Batch upsert
If you have a large number of vectors, you can upsert them in batches.
import asyncio from pinecone import Pinecone, Vector, SparseValues async def main(): pc = Pinecone() async with pc.IndexAsyncio(host="example-dojoi3u.svc.aped-4627-b74a.pinecone.io") as idx: await idx.upsert( namespace = 'my-namespace', vectors = [ {'id': 'id1', 'values': [0.1, 0.2, 0.3, 0.4]}, {'id': 'id2', 'values': [0.2, 0.3, 0.4, 0.5]}, {'id': 'id3', 'values': [0.3, 0.4, 0.5, 0.6]}, {'id': 'id4', 'values': [0.4, 0.5, 0.6, 0.7]}, {'id': 'id5', 'values': [0.5, 0.6, 0.7, 0.8]}, # More vectors here ], batch_size = 50 ) asyncio.run(main())
Visual progress bar with tqdm
To see a progress bar when upserting in batches, you will need to separately install the tqdm package. If tqdm is present, the client will detect and use it to display progress when show_progress=True.
- IndexAsyncio.query(*args, top_k: int, vector: List[float] | None = None, id: str | None = None, namespace: str | None = None, filter: Dict[str, str | int | float | bool] | Dict[Literal['$eq'], str | int | float | bool] | Dict[Literal['$ne'], str | int | float | bool] | Dict[Literal['$gt'], int | float] | Dict[Literal['$gte'], int | float] | Dict[Literal['$lt'], int | float] | Dict[Literal['$lte'], int | float] | Dict[Literal['$in'], List[str | int | float | bool]] | Dict[Literal['$nin'], List[str | int | float | bool]] | Dict[Literal['$and'], List[Dict[str, str | int | float | bool] | Dict[Literal['$eq'], str | int | float | bool] | Dict[Literal['$ne'], str | int | float | bool] | Dict[Literal['$gt'], int | float] | Dict[Literal['$gte'], int | float] | Dict[Literal['$lt'], int | float] | Dict[Literal['$lte'], int | float] | Dict[Literal['$in'], List[str | int | float | bool]] | Dict[Literal['$nin'], List[str | int | float | bool]]]] | None = None, include_values: bool | None = None, include_metadata: bool | None = None, sparse_vector: SparseValues | SparseVectorTypedDict | None = None, **kwargs) QueryResponse ¶
The Query operation searches a namespace, using a query vector. It retrieves the ids of the most similar items in a namespace, along with their similarity scores.
Querying with dense vectors
import asyncio from pinecone import Pinecone, Vector, SparseValues async def main(): pc = Pinecone() async with pc.IndexAsyncio(host="example-dojoi3u.svc.aped-4627-b74a.pinecone.io") as idx: query_embedding = [0.1, 0.2, 0.3, ...] # An embedding that matches the index dimension # Query by vector values results = await idx.query( vector=query_embedding, top_k=10, filter={'genre': {"$eq": "drama"}}, # Optionally filter by metadata namespace='my_namespace', include_values=False, include_metadata=True ) # Query using vector id (the values from this stored vector will be used to query) results = await idx.query( id='1', top_k=10, filter={"year": {"$gt": 2000}}, namespace='my_namespace', ) asyncio.run(main())
Query with sparse vectors
import asyncio from pinecone import Pinecone, Vector, SparseValues async def main(): pc = Pinecone() async with pc.IndexAsyncio(host="example-dojoi3u.svc.aped-4627-b74a.pinecone.io") as idx: query_embedding = [0.1, 0.2, 0.3, ...] # An embedding that matches the index dimension # Query by vector values results = await idx.query( vector=query_embedding, top_k=10, filter={'genre': {"$eq": "drama"}}, # Optionally filter by metadata namespace='my_namespace', include_values=False, include_metadata=True ) # Query using vector id (the values from this stored vector will be used to query) results = await idx.query( id='1', top_k=10, filter={"year": {"$gt": 2000}}, namespace='my_namespace', ) asyncio.run(main())
Examples:
>>> index.query(vector=[1, 2, 3], top_k=10, namespace='my_namespace') >>> index.query(id='id1', top_k=10, namespace='my_namespace') >>> index.query(vector=[1, 2, 3], top_k=10, namespace='my_namespace', filter={'key': 'value'}) >>> index.query(id='id1', top_k=10, namespace='my_namespace', include_metadata=True, include_values=True) >>> index.query(vector=[1, 2, 3], sparse_vector={'indices': [1, 2], 'values': [0.2, 0.4]}, >>> top_k=10, namespace='my_namespace') >>> index.query(vector=[1, 2, 3], sparse_vector=SparseValues([1, 2], [0.2, 0.4]), >>> top_k=10, namespace='my_namespace')
- Parameters:
vector (List[float]) – The query vector. This should be the same length as the dimension of the index being queried. Each query() request can contain only one of the parameters id or vector.. [optional]
id (str) – The unique ID of the vector to be used as a query vector. Each query() request can contain only one of the parameters vector or id. [optional]
top_k (int) – The number of results to return for each query. Must be an integer greater than 1.
namespace (str) – The namespace to fetch vectors from. If not specified, the default namespace is used. [optional]
filter (Dict[str, Union[str, float, int, bool, List, dict]) – The filter to apply. You can use vector metadata to limit your search. See metadata filtering <https://www.pinecone.io/docs/metadata-filtering/>_ [optional]
include_values (bool) – Indicates whether vector values are included in the response. If omitted the server will use the default value of False [optional]
include_metadata (bool) – Indicates whether metadata is included in the response as well as the ids. If omitted the server will use the default value of False [optional]
sparse_vector –
(Union[SparseValues, Dict[str, Union[List[float], List[int]]]]): sparse values of the query vector. Expected to be either a SparseValues object or a dict of the form:
{‘indices’: List[int], ‘values’: List[float]}, where the lists each have the same length.
- Returns: QueryResponse object which contains the list of the closest vectors as ScoredVector objects,
and namespace name.
- IndexAsyncio.query_namespaces(namespaces: List[str], metric: Literal['cosine', 'euclidean', 'dotproduct'], top_k: int | None = None, filter: Dict[str, str | float | int | bool | List | dict] | None = None, include_values: bool | None = None, include_metadata: bool | None = None, vector: List[float] | None = None, sparse_vector: SparseValues | Dict[str, List[float] | List[int]] | None = None, **kwargs) QueryNamespacesResults ¶
The query_namespaces() method is used to make a query to multiple namespaces in parallel and combine the results into one result set.
- Parameters:
vector (List[float]) – The query vector, must be the same length as the dimension of the index being queried.
namespaces (List[str]) – The list of namespaces to query.
top_k (Optional[int], optional) – The number of results you would like to request from each namespace. Defaults to 10.
filter (Optional[Dict[str, Union[str, float, int, bool, List, dict]]], optional) – Pass an optional filter to filter results based on metadata. Defaults to None.
include_values (Optional[bool], optional) – Boolean field indicating whether vector values should be included with results. Defaults to None.
include_metadata (Optional[bool], optional) – Boolean field indicating whether vector metadata should be included with results. Defaults to None.
sparse_vector (Optional[ Union[SparseValues, Dict[str, Union[List[float], List[int]]]] ], optional) – If you are working with a dotproduct index, you can pass a sparse vector as part of your hybrid search. Defaults to None.
- Returns:
A QueryNamespacesResults object containing the combined results from all namespaces, as well as the combined usage cost in read units.
- Return type:
QueryNamespacesResults
Examples:
import asyncio from pinecone import Pinecone async def main(): pc = Pinecone(api_key="your-api-key") idx = pc.IndexAsyncio( host="example-dojoi3u.svc.aped-4627-b74a.pinecone.io", ) query_vec = [0.1, 0.2, 0.3] # An embedding that matches the index dimension combined_results = await idx.query_namespaces( vector=query_vec, namespaces=['ns1', 'ns2', 'ns3', 'ns4'], top_k=10, filter={'genre': {"$eq": "drama"}}, include_values=True, include_metadata=True ) for vec in combined_results.matches: print(vec.id, vec.score) print(combined_results.usage) await idx.close() asyncio.run(main())
- IndexAsyncio.delete(ids: List[str] | None = None, delete_all: bool | None = None, namespace: str | None = None, filter: Dict[str, str | int | float | bool] | Dict[Literal['$eq'], str | int | float | bool] | Dict[Literal['$ne'], str | int | float | bool] | Dict[Literal['$gt'], int | float] | Dict[Literal['$gte'], int | float] | Dict[Literal['$lt'], int | float] | Dict[Literal['$lte'], int | float] | Dict[Literal['$in'], List[str | int | float | bool]] | Dict[Literal['$nin'], List[str | int | float | bool]] | Dict[Literal['$and'], List[Dict[str, str | int | float | bool] | Dict[Literal['$eq'], str | int | float | bool] | Dict[Literal['$ne'], str | int | float | bool] | Dict[Literal['$gt'], int | float] | Dict[Literal['$gte'], int | float] | Dict[Literal['$lt'], int | float] | Dict[Literal['$lte'], int | float] | Dict[Literal['$in'], List[str | int | float | bool]] | Dict[Literal['$nin'], List[str | int | float | bool]]]] | None = None, **kwargs) Dict[str, Any] ¶
- Parameters:
ids (List[str]) – Vector ids to delete [optional]
delete_all (bool) – This indicates that all vectors in the index namespace should be deleted.. [optional] Default is False.
namespace (str) – The namespace to delete vectors from [optional] If not specified, the default namespace is used.
filter (Dict[str, Union[str, float, int, bool, List, dict]]) – If specified, the metadata filter here will be used to select the vectors to delete. This is mutually exclusive with specifying ids to delete in the ids param or using delete_all=True. See metadata filtering <https://www.pinecone.io/docs/metadata-filtering/>_ [optional]
The Delete operation deletes vectors from the index, from a single namespace.
No error is raised if the vector id does not exist.
Note: For any delete call, if namespace is not specified, the default namespace “” is used. Since the delete operation does not error when ids are not present, this means you may not receive an error if you delete from the wrong namespace.
Delete can occur in the following mutual exclusive ways:
Delete by ids from a single namespace
Delete all vectors from a single namespace by setting delete_all to True
- Delete all vectors from a single namespace by specifying a metadata filter
(note that for this option delete all must be set to False)
import asyncio from pinecone import Pinecone, Vector, SparseValues async def main(): pc = Pinecone() async with pc.IndexAsyncio(host="example-dojoi3u.svc.aped-4627-b74a.pinecone.io") as idx: # Delete specific ids await idx.delete( ids=['id1', 'id2'], namespace='my_namespace' ) # Delete everything in a namespace await idx.delete( delete_all=True, namespace='my_namespace' ) # Delete by metadata filter await idx.delete( filter={'key': 'value'}, namespace='my_namespace' ) asyncio.run(main())
Returns: An empty dictionary if the delete operation was successful.
- IndexAsyncio.fetch(ids: List[str], namespace: str | None = None, **kwargs) FetchResponse ¶
The fetch operation looks up and returns vectors, by ID, from a single namespace. The returned vectors include the vector data and/or metadata.
import asyncio from pinecone import Pinecone, Vector, SparseValues async def main(): pc = Pinecone() async with pc.IndexAsyncio(host="example-dojoi3u.svc.aped-4627-b74a.pinecone.io") as idx: # Fetch specific ids in namespace fetched = await idx.fetch( ids=['id1', 'id2'], namespace='my_namespace' ) for vec_id in fetched.vectors: vector = fetched.vectors[vec_id] print(vector.id) print(vector.metadata) print(vector.values) asyncio.run(main())
- Parameters:
ids (List[str]) – The vector IDs to fetch.
namespace (str) – The namespace to fetch vectors from. If not specified, the default namespace is used. [optional]
Returns: FetchResponse object which contains the list of Vector objects, and namespace name.
- IndexAsyncio.list(**kwargs)¶
The list operation accepts all of the same arguments as list_paginated, and returns a generator that yields a list of the matching vector ids in each page of results. It automatically handles pagination tokens on your behalf.
Examples
>>> for ids in index.list(prefix='99', limit=5, namespace='my_namespace'): >>> print(ids) ['99', '990', '991', '992', '993'] ['994', '995', '996', '997', '998'] ['999']
- Parameters:
prefix (Optional[str]) – The id prefix to match. If unspecified, an empty string prefix will be used with the effect of listing all ids in a namespace [optional]
limit (Optional[int]) – The maximum number of ids to return. If unspecified, the server will use a default value. [optional]
pagination_token (Optional[str]) – A token needed to fetch the next page of results. This token is returned in the response if additional results are available. [optional]
namespace (Optional[str]) – The namespace to fetch vectors from. If not specified, the default namespace is used. [optional]
- IndexAsyncio.list_paginated(prefix: str | None = None, limit: int | None = None, pagination_token: str | None = None, namespace: str | None = None, **kwargs) ListResponse ¶
The list_paginated operation finds vectors based on an id prefix within a single namespace. It returns matching ids in a paginated form, with a pagination token to fetch the next page of results. This id list can then be passed to fetch or delete operations, depending on your use case.
Consider using the list method to avoid having to handle pagination tokens manually.
Examples
>>> results = index.list_paginated(prefix='99', limit=5, namespace='my_namespace') >>> [v.id for v in results.vectors] ['99', '990', '991', '992', '993'] >>> results.pagination.next eyJza2lwX3Bhc3QiOiI5OTMiLCJwcmVmaXgiOiI5OSJ9 >>> next_results = index.list_paginated(prefix='99', limit=5, namespace='my_namespace', pagination_token=results.pagination.next)
- Parameters:
prefix (Optional[str]) – The id prefix to match. If unspecified, an empty string prefix will be used with the effect of listing all ids in a namespace [optional]
limit (Optional[int]) – The maximum number of ids to return. If unspecified, the server will use a default value. [optional]
pagination_token (Optional[str]) – A token needed to fetch the next page of results. This token is returned in the response if additional results are available. [optional]
namespace (Optional[str]) – The namespace to fetch vectors from. If not specified, the default namespace is used. [optional]
Returns: ListResponse object which contains the list of ids, the namespace name, pagination information, and usage showing the number of read_units consumed.
Records¶
If you have created an index using integrated inference, you can use the following methods to search and retrieve records.
- async IndexAsyncio.search(namespace: str, query: SearchQueryTypedDict | SearchQuery, rerank: SearchRerankTypedDict | SearchRerank | None = None, fields: List[str] | None = ['*']) SearchRecordsResponse ¶
- Parameters:
namespace (str, required) – The namespace in the index to search.
query (Union[Dict, SearchQuery], required) – The SearchQuery to use for the search.
rerank (Union[Dict, SearchRerank], optional) – The SearchRerank to use with the search request.
- Returns:
The records that match the search.
Search for records.
This operation converts a query to a vector embedding and then searches a namespace. You can optionally provide a reranking operation as part of the search.
import asyncio from pinecone import ( Pinecone, CloudProvider, AwsRegion, EmbedModel IndexEmbed ) async def main(): pc = Pinecone() async with pc.IndexAsyncio(host="example-dojoi3u.svc.aped-4627-b74a.pinecone.io") as idx: # upsert records await idx.upsert_records( namespace="my-namespace", records=[ { "_id": "test1", "my_text_field": "Apple is a popular fruit known for its sweetness and crisp texture.", }, { "_id": "test2", "my_text_field": "The tech company Apple is known for its innovative products like the iPhone.", }, { "_id": "test3", "my_text_field": "Many people enjoy eating apples as a healthy snack.", }, { "_id": "test4", "my_text_field": "Apple Inc. has revolutionized the tech industry with its sleek designs and user-friendly interfaces.", }, { "_id": "test5", "my_text_field": "An apple a day keeps the doctor away, as the saying goes.", }, { "_id": "test6", "my_text_field": "Apple Computer Company was founded on April 1, 1976, by Steve Jobs, Steve Wozniak, and Ronald Wayne as a partnership.", }, ], ) from pinecone import SearchQuery, SearchRerank, RerankModel # search for similar records response = await idx.search_records( namespace="my-namespace", query=SearchQuery( inputs={ "text": "Apple corporation", }, top_k=3, ), rerank=SearchRerank( model=RerankModel.Bge_Reranker_V2_M3, rank_fields=["my_text_field"], top_n=3, ), ) asyncio.run(main())
- async IndexAsyncio.search_records(namespace: str, query: SearchQueryTypedDict | SearchQuery, rerank: SearchRerankTypedDict | SearchRerank | None = None, fields: List[str] | None = ['*']) SearchRecordsResponse ¶
Alias of the search() method.
Inference¶
- Inference.embed(model: EmbedModel | str, inputs: str | List[Dict] | List[str], parameters: Dict[str, Any] | None = None) EmbeddingsList [source]¶
Generates embeddings for the provided inputs using the specified model and (optional) parameters.
- Parameters:
model (str, required) – The model to use for generating embeddings.
inputs (list, required) – A list of items to generate embeddings for.
parameters (dict, optional) – A dictionary of parameters to use when generating embeddings.
- Returns:
EmbeddingsList
object with keysdata
,model
, andusage
. Thedata
key contains a list ofn
embeddings, wheren
= len(inputs). Precision of returned embeddings is either float16 or float32, with float32 being the default.model
key is the model used to generate the embeddings.usage
key contains the total number of tokens used at request-time.
Example:
>>> pc = Pinecone() >>> inputs = ["Who created the first computer?"] >>> outputs = pc.inference.embed(model="multilingual-e5-large", inputs=inputs, parameters={"input_type": "passage", "truncate": "END"}) >>> print(outputs) EmbeddingsList( model='multilingual-e5-large', data=[ {'values': [0.1, ...., 0.2]}, ], usage={'total_tokens': 6} )
- Inference.rerank(model: RerankModel | str, query: str, documents: List[str] | List[Dict[str, Any]], rank_fields: List[str] = ['text'], return_documents: bool = True, top_n: int | None = None, parameters: Dict[str, Any] | None = None) RerankResult [source]¶
Rerank documents with associated relevance scores that represent the relevance of each document to the provided query using the specified model.
- Parameters:
model (str, required) – The model to use for reranking.
query (str, required) – The query to compare with documents.
documents (list, required) – A list of documents or strings to rank.
rank_fields (list, optional) – A list of document fields to use for ranking. Defaults to [“text”].
return_documents (bool, optional) – Whether to include the documents in the response. Defaults to True.
top_n (int, optional) – How many documents to return. Defaults to len(documents).
parameters (dict, optional) – A dictionary of parameters to use when ranking documents.
- Returns:
RerankResult
object with keysdata
andusage
. Thedata
key contains a list ofn
documents, wheren
=top_n
and type(n) = Document. The documents are sorted in order of relevance, with the first being the most relevant. Theindex
field can be used to locate the document relative to the list of documents specified in the request. Each document contains ascore
key representing how close the document relates to the query.
Example:
>>> pc = Pinecone() >>> pc.inference.rerank( model="bge-reranker-v2-m3", query="Tell me about tech companies", documents=[ "Apple is a popular fruit known for its sweetness and crisp texture.", "Software is still eating the world.", "Many people enjoy eating apples as a healthy snack.", "Acme Inc. has revolutionized the tech industry with its sleek designs and user-friendly interfaces.", "An apple a day keeps the doctor away, as the saying goes.", ], top_n=2, return_documents=True, ) RerankResult( model='bge-reranker-v2-m3', data=[{ index=3, score=0.020924192, document={ text='Acme Inc. has revolutionized the tech industry with its sleek designs and user-friendly interfaces.' } },{ index=1, score=0.00034464317, document={ text='Software is still eating the world.' } }], usage={'rerank_units': 1} )
- Inference.list_models(*, type: str | None = None, vector_type: str | None = None) ModelInfoList [source]¶
List all available models.
- Parameters:
type (str, optional) – The type of model to list. Either “embed” or “rerank”.
vector_type (str, optional) – The type of vector to list. Either “dense” or “sparse”.
- Returns:
A list of models.
Example:
pc = Pinecone() # List all models models = pc.inference.list_models() # List models, with model type filtering models = pc.inference.list_models(type="embed") models = pc.inference.list_models(type="rerank") # List models, with vector type filtering models = pc.inference.list_models(vector_type="dense") models = pc.inference.list_models(vector_type="sparse") # List models, with both type and vector type filtering models = pc.inference.list_models(type="rerank", vector_type="dense")
- Inference.get_model(model_name: str) ModelInfo [source]¶
Get details on a specific model.
- Parameters:
model_name (str, required) – The name of the model to get details on.
- Returns:
A ModelInfo object.
>>> pc = Pinecone() >>> pc.inference.get_model(model_name="pinecone-rerank-v0") { "model": "pinecone-rerank-v0", "short_description": "A state of the art reranking model that out-performs competitors on widely accepted benchmarks. It can handle chunks up to 512 tokens (1-2 paragraphs)", "type": "rerank", "supported_parameters": [ { "parameter": "truncate", "type": "one_of", "value_type": "string", "required": false, "default": "END", "allowed_values": [ "END", "NONE" ] } ], "modality": "text", "max_sequence_length": 512, "max_batch_size": 100, "provider_name": "Pinecone", "supported_metrics": [] }