### Complete Workflow Example Source: https://github.com/volcengine/vikingdb-python-sdk/blob/main/_autodocs/README.md Demonstrates a complete workflow including authentication, client initialization, and getting an index. Uses environment variables for sensitive information. ```python import os from vikingdb import IAM, VikingDB, RequestOptions from vikingdb.vector import SearchByVectorRequest, EmbeddingRequest # 1. Authentication auth = IAM( ak=os.environ["VIKINGDB_AK"], sk=os.environ["VIKINGDB_SK"], ) # 2. Initialize clients vector_client = VikingDB( host=os.environ["VIKINGDB_HOST"], region=os.environ["VIKINGDB_REGION"], auth=auth, ) # 3. Get index index = vector_client.index( collection_name="my_collection", index_name="my_index", ) ``` -------------------------------- ### Search Knowledge Example Source: https://github.com/volcengine/vikingdb-python-sdk/blob/main/_autodocs/api-reference/vikingknowledge-client.md Example demonstrating a knowledge search with query, limit, and dense_weight for retrieval balancing. ```python from vikingdb.knowledge import SearchKnowledgeRequest response = collection.search_knowledge( SearchKnowledgeRequest( query="2025 Q1 revenue", limit=5, dense_weight=0.5, # Balance dense/sparse search ) ) ``` -------------------------------- ### Install Pytest Source: https://github.com/volcengine/vikingdb-python-sdk/blob/main/README.md Install pytest using uv for development dependencies. ```bash uv add --dev pytest ``` -------------------------------- ### Run Knowledge Examples Source: https://github.com/volcengine/vikingdb-python-sdk/blob/main/README.md Execute individual knowledge base management scripts. ```bash python examples/knowledge/01_init_client.py python examples/knowledge/02_doc_crud.py python examples/knowledge/03_point_crud.py python examples/knowledge/04_search.py ``` -------------------------------- ### Run Memory Examples Source: https://github.com/volcengine/vikingdb-python-sdk/blob/main/README.md Execute individual memory management scripts. ```bash python examples/memory/01_init_client_and_collection.py python examples/memory/02_add_session.py python examples/memory/03_search_memory.py ``` -------------------------------- ### API Key Authentication Setup Source: https://github.com/volcengine/vikingdb-python-sdk/blob/main/_autodocs/configuration.md Configure authentication using an API key token. ```python from vikingdb import APIKey auth = APIKey(api_key="your-api-key") ``` -------------------------------- ### Complete Knowledge Workflow Initialization Source: https://github.com/volcengine/vikingdb-python-sdk/blob/main/_autodocs/api-reference/vikingknowledge-client.md Demonstrates the initialization of the VikingKnowledge client with IAM authentication and collection setup. ```python from vikingdb import VikingKnowledge, IAM from vikingdb.knowledge import ( AddDocV2Request, AddPointRequest, SearchKnowledgeRequest, ChatCompletionRequest, ChatMessage, ) # Initialize auth = IAM(ak="ak", sk="sk") client = VikingKnowledge( host="api-knowledgebase.mlp.cn-beijing.volces.com", region="cn-beijing", auth=auth, ) collection = client.collection( collection_name="financial", project_name="default", ) # 1. Add document doc_response = collection.add_doc_v2( AddDocV2Request( doc_name="Q1 2025 Report", doc_type="pdf", uri="tos://bucket/q1_2025.pdf", ) ) doc_id = doc_response.result.doc_id ``` -------------------------------- ### Set Environment Variables for Memory Examples Source: https://github.com/volcengine/vikingdb-python-sdk/blob/main/README.md Configure access keys for memory operations. ```bash export VIKINGDB_AK=your-access-key export VIKINGDB_SK=your-secret-key ``` -------------------------------- ### API Key Authentication Example Source: https://github.com/volcengine/vikingdb-python-sdk/blob/main/_autodocs/api-reference/auth.md Initializes an APIKey authentication provider and creates a VikingKnowledge client. ```python from vikingdb import APIKey, VikingKnowledge auth = APIKey(api_key="your-api-key") client = VikingKnowledge( host="api-knowledgebase.mlp.cn-beijing.volces.com", region="cn-beijing", auth=auth, ) ``` -------------------------------- ### Chat Completion Example Source: https://github.com/volcengine/vikingdb-python-sdk/blob/main/_autodocs/api-reference/vikingknowledge-client.md Example of initiating a chat completion with a user message, specifying a model and retrieval options. ```python from vikingdb.knowledge import ChatCompletionRequest, ChatMessage response = collection.chat_completion( ChatCompletionRequest( messages=[ ChatMessage(role="user", content="What was Q1 revenue?"), ], model="gpt-4", retrieve_options={"limit": 5}, ) ) print(response.result.choices[0].message.content) ``` -------------------------------- ### Complete Memory Operation with Exception Handling and Retries Source: https://github.com/volcengine/vikingdb-python-sdk/blob/main/_autodocs/api-reference/memory-exceptions.md A comprehensive example showing how to initialize the VikingMem client, get a collection, add an event, and handle various exceptions including `UnauthorizedException`, `CollectionNotExistException`, and `QuotaLimiterException` with retries. ```python from vikingdb import IAM from vikingdb.memory import ( VikingMem, VikingMemException, UnauthorizedException, CollectionNotExistException, QuotaLimiterException, ) import time def safe_memory_operation(): # Initialize with error handling try: auth = IAM(ak="ak", sk="sk") client = VikingMem( host="api-knowledgebase.mlp.cn-beijing.volces.com", region="cn-beijing", auth=auth, ) if not client.ping(): print("Service not available") return except UnauthorizedException: print("Invalid credentials") return # Get collection try: collection = client.get_collection( collection_name="demo", ) except CollectionNotExistException: print("Collection not found") return # Add event with retry max_retries = 3 for attempt in range(1, max_retries + 1): try: result = collection.add_event( event_type="test", memory_info={"key": "value"}, user_id="user_001", ) print("Event added successfully") return except QuotaLimiterException: if attempt < max_retries: delay = 2 ** (attempt - 1) print(f"Rate limited, retrying in {delay}s") time.sleep(delay) continue print("Rate limited, max retries exceeded") return except VikingMemException as e: print(f"Error {e.code}: {e.message}") return safe_memory_operation() ``` -------------------------------- ### Header Authentication Example Source: https://github.com/volcengine/vikingdb-python-sdk/blob/main/_autodocs/api-reference/auth.md Initializes a HeaderAuth provider with custom headers and creates a VikingDB client. ```python from vikingdb import HeaderAuth, VikingDB auth = HeaderAuth(headers={ "X-Custom-Auth": "custom-value", "X-API-Key": "key123" }) client = VikingDB( host="api-vikingdb.vikingdb.ap-southeast-1.volces.com", region="ap-southeast-1", auth=auth, ) ``` -------------------------------- ### Search Collection Example Source: https://github.com/volcengine/vikingdb-python-sdk/blob/main/_autodocs/api-reference/vikingknowledge-client.md Example of how to perform a collection search using SearchCollectionRequest and iterate through results. ```python from vikingdb.knowledge import SearchCollectionRequest response = collection.search_collection( SearchCollectionRequest( query="machine learning algorithms", limit=10, ) ) for result in response.result.result_list: print(f"Document: {result.doc_id}, Score: {result.score}") ``` -------------------------------- ### IAM Authentication Example Source: https://github.com/volcengine/vikingdb-python-sdk/blob/main/_autodocs/api-reference/auth.md Initializes an IAM authentication provider with access key and secret key, then creates a VikingDB client. ```python from vikingdb import IAM, VikingDB auth = IAM(ak="your-access-key", sk="your-secret-key") client = VikingDB( host="api-vikingdb.vikingdb.ap-southeast-1.volces.com", region="ap-southeast-1", auth=auth, ) ``` -------------------------------- ### Install VikingDB Python SDK Source: https://github.com/volcengine/vikingdb-python-sdk/blob/main/README.md Install the VikingDB Python SDK using uv. Ensure you have the necessary dependencies like requests and pydantic. ```bash uv add vikingdb-python-sdk ``` -------------------------------- ### Example Usage of SearchByVectorRequest Source: https://github.com/volcengine/vikingdb-python-sdk/blob/main/_autodocs/api-reference/vector-types.md Demonstrates how to construct a SearchByVectorRequest to find items similar to a given dense vector. This example includes setting a limit, specifying output fields, applying a filter, and configuring advanced search options like dense weight and specific IDs. ```python from vikingdb.vector import SearchByVectorRequest request = SearchByVectorRequest( dense_vector=[0.1, 0.2, 0.3, 0.4], limit=10, output_fields=["id", "text", "score"], filter={"category": "electronics"}, advance=SearchAdvance( dense_weight=1.0, ids_in=["id1", "id2"], ), ) response = index.search_by_vector(request) ``` -------------------------------- ### Set Environment Variables for Knowledge Examples Source: https://github.com/volcengine/vikingdb-python-sdk/blob/main/README.md Configure access keys, project, and collection details for knowledge base operations. API keys for chat features and optional rerank instructions can also be set. ```bash VOLC_AK=your-access-key VOLC_SK=your-secret-key VIKING_PROJECT=project-name VIKING_COLLECTION_RID=collection-resource-id VIKING_COLLECTION_NAME=collection-name # For chat/service features: VIKING_CHAT_API_KEY=chat-api-key VIKINGDB_SERVICE_API_KEY=service-api-key VIKINGDB_SERVICE_RID=service-resource-id # Optional for rerank: VIKINGDB_RERANK_INSTRUCTION=custom-instruction ``` -------------------------------- ### IAM Authentication Setup Source: https://github.com/volcengine/vikingdb-python-sdk/blob/main/_autodocs/configuration.md Configure IAM authentication using Access Key and Secret Key. Environment variables can also be used. ```python from vikingdb import IAM auth = IAM(ak="your-access-key", sk="your-secret-key") ``` ```python import os from vikingdb import IAM, VikingDB auth = IAM( ak=os.getenv("VIKINGDB_AK"), sk=os.getenv("VIKINGDB_SK"), ) client = VikingDB( host=os.getenv("VIKINGDB_HOST"), region=os.getenv("VIKINGDB_REGION"), auth=auth, ) ``` -------------------------------- ### Complete Search Workflow Example Source: https://github.com/volcengine/vikingdb-python-sdk/blob/main/_autodocs/api-reference/index-client.md This snippet shows how to initialize the VikingDB client, access an index, and perform vector, keyword, and scalar searches, along with an aggregation operation. ```python from vikingdb import VikingDB, IAM from vikingdb.vector import ( SearchByVectorRequest, SearchByKeywordsRequest, SearchByScalarRequest, ) auth = IAM(ak="ak", sk="sk") client = VikingDB( host="api-vikingdb.vikingdb.ap- vikingdb.ap-southeast-1.volces.com", region="ap-southeast-1", auth=auth, ) index = client.index( collection_name="products", index_name="price_index", ) # 1. Vector search (semantic) vector_results = index.search_by_vector( SearchByVectorRequest( dense_vector=[0.1, 0.2, 0.3], limit=5, ) ) # 2. Keyword search (full-text) keyword_results = index.search_by_keywords( SearchByKeywordsRequest( query="laptop computer", limit=5, ) ) # 3. Scalar search (sorting) scalar_results = index.search_by_scalar( SearchByScalarRequest( field="price", order="asc", limit=5, ) ) # 4. Aggregation count_result = index.aggregate( AggRequest(op="count") ) ``` -------------------------------- ### Handle Authentication Failure Source: https://github.com/volcengine/vikingdb-python-sdk/blob/main/_autodocs/api-reference/memory-exceptions.md Example of how to catch and handle an UnauthorizedException when authentication credentials are invalid. ```python from vikingdb.memory import VikingMem, UnauthorizedException, IAM try: auth = IAM(ak="invalid-ak", sk="invalid-sk") client = VikingMem( host="api-knowledgebase.mlp.cn-beijing.volces.com", region="cn-beijing", auth=auth, ) client.ping() except UnauthorizedException as e: print("Authentication failed") print("Verify AK/SK or API key credentials") ``` -------------------------------- ### Header-Based Authentication Setup Source: https://github.com/volcengine/vikingdb-python-sdk/blob/main/_autodocs/configuration.md Configure custom header-based authentication, intended for internal or testing purposes. ```python from vikingdb import HeaderAuth auth = HeaderAuth(headers={ "X-Custom-Auth": "value", "X-API-Key": "key123", }) ``` -------------------------------- ### Handle QuotaLimiterException Source: https://github.com/volcengine/vikingdb-python-sdk/blob/main/_autodocs/api-reference/knowledge-exceptions.md Example of catching QuotaLimiterException and implementing a backoff strategy. ```python from vikingdb.knowledge import QuotaLimiterException import time try: for i in range(100): collection.add_doc_v2(request) except QuotaLimiterException as e: print("Rate limited, backing off") time.sleep(10) ``` -------------------------------- ### collection Source: https://github.com/volcengine/vikingdb-python-sdk/blob/main/_autodocs/api-reference/vikingknowledge-client.md Get a collection instance for document and point operations. ```APIDOC ## collection Get a collection instance for document and point operations. ### Description Get a collection instance for document and point operations. ### Method None (This is a method call on the client instance) ### Endpoint None (This is a method call on the client instance) ### Parameters #### Path Parameters None #### Query Parameters None #### Request Body None ### Method Signature ```python def collection( self, *, resource_id: Optional[str] = None, collection_name: Optional[str] = None, project_name: Optional[str] = None, ) -> KnowledgeCollection ``` | Parameter | Type | Required | Description | |-----------|------|----------|-------------| | resource_id | Optional[str] | No | Collection resource ID | | collection_name | Optional[str] | No | Collection name | | project_name | Optional[str] | No | Project name | **Returns:** `KnowledgeCollection` instance **Example:** ```python from vikingdb import VikingKnowledge, IAM client = VikingKnowledge( host="api-knowledgebase.mlp.cn-beijing.volces.com", region="cn-beijing", auth=IAM(ak="ak", sk="sk"), ) collection = client.collection( resource_id="collection-rid-123", ) # Or by name collection = client.collection( collection_name="financial_reports", project_name="default", ) ``` ``` -------------------------------- ### Basic Re-ranking Example Source: https://github.com/volcengine/vikingdb-python-sdk/blob/main/_autodocs/api-reference/rerank-client.md Demonstrates how to initialize the VikingDB client and use the `rerank_client.rerank` method with a `RerankRequest` containing specific items. Accesses and prints the re-ranked results. ```python from vikingdb import VikingDB, IAM from vikingdb.vector import RerankRequest, Rerank client = VikingDB( host="api-vikingdb.vikingdb.ap-southeast-1.volces.com", region="ap-southeast-1", auth=IAM(ak="ak", sk="sk"), ) rerank_client = client.rerank() # Re-rank search results response = rerank_client.rerank( RerankRequest( query="best laptop for programming", model="bge-reranker-large", items=[ Rerank(index=0, text="MacBook Pro with 16GB RAM", score=0.92), Rerank(index=1, text="Dell XPS 13 ultrabook", score=0.89), Rerank(index=2, text="ThinkPad X1 Carbon", score=0.87), Rerank(index=3, text="ASUS VivoBook", score=0.75), ], ) ) # Access re-ranked results for i, item in enumerate(response.result.data): print(f"Rank {i+1}: {item}") ``` -------------------------------- ### Execute Vector Scenario Tests Source: https://github.com/volcengine/vikingdb-python-sdk/blob/main/README.md Run pytest scenarios for vector examples using uv. ```bash uv run pytest examples/vector -k scenario ``` -------------------------------- ### Handle DocIsFullException Source: https://github.com/volcengine/vikingdb-python-sdk/blob/main/_autodocs/api-reference/knowledge-exceptions.md Example of catching DocIsFullException when attempting to add a point to a full document. ```python from vikingdb.knowledge import DocIsFullException try: response = collection.add_point( AddPointRequest(doc_id="doc_123", content="...") ) except DocIsFullException as e: print("Document is full, cannot add more points") ``` -------------------------------- ### Handle IndexNotReadyException Source: https://github.com/volcengine/vikingdb-python-sdk/blob/main/_autodocs/api-reference/knowledge-exceptions.md Example of catching IndexNotReadyException and retrying the search after a delay. ```python from vikingdb.knowledge import IndexNotReadyException import time try: response = collection.search_knowledge(query="test") except IndexNotReadyException as e: print("Index not ready, waiting...") time.sleep(5) response = collection.search_knowledge(query="test") ``` -------------------------------- ### Set Environment Variables for Vector Examples Source: https://github.com/volcengine/vikingdb-python-sdk/blob/main/README.md Configure access keys, collection, and index names for vector operations. Optional project and resource IDs can also be set. ```bash VIKINGDB_AK=your-access-key VIKINGDB_SK=your-secret-key VIKINGDB_COLLECTION=demo_collection VIKINGDB_INDEX=demo_index # Optional: # VIKINGDB_PROJECT=project-name # VIKINGDB_RESOURCE_ID=resource-id ``` -------------------------------- ### Comprehensive Knowledge Workflow with Error Handling and Retries Source: https://github.com/volcengine/vikingdb-python-sdk/blob/main/_autodocs/api-reference/knowledge-exceptions.md A complete example demonstrating initialization, collection access, document and point addition (with retries for IndexNotReadyException), and searching, all wrapped in comprehensive error handling for various VikingKnowledge exceptions. ```python from vikingdb import IAM from vikingdb.knowledge import ( VikingKnowledge, VikingKnowledgeException, UnauthorizedException, CollectionNotExistException, DocIsFullException, IndexNotReadyException, AddDocV2Request, AddPointRequest, SearchKnowledgeRequest, ) import time def safe_knowledge_workflow(): # 1. Initialize with error handling try: auth = IAM(ak="ak", sk="sk") client = VikingKnowledge( host="api-knowledgebase.mlp.cn-beijing.volces.com", region="cn-beijing", auth=auth, ) except UnauthorizedException: print("Invalid credentials") return # 2. Get collection try: collection = client.collection( collection_name="financial_docs", ) except CollectionNotExistException: print("Collection not found") return # 3. Add document try: doc_response = collection.add_doc_v2( AddDocV2Request( doc_name="Q1 2025 Report", doc_type="pdf", uri="tos://bucket/q1_2025.pdf", ) ) doc_id = doc_response.result.doc_id except VikingKnowledgeException as e: print(f"Failed to add document: {e.code} - {e.message}") return # 4. Add points (with retry for IndexNotReadyException) max_retries = 3 for attempt in range(1, max_retries + 1): try: point_response = collection.add_point( AddPointRequest( doc_id=doc_id, content="Revenue grew 15% YoY...", ) ) print("Point added successfully") break except IndexNotReadyException: if attempt < max_retries: print("Index not ready, waiting...") time.sleep(5) continue print("Index not ready after retries") return except DocIsFullException: print("Document is full, cannot add more points") return except VikingKnowledgeException as e: print(f"Error {e.code}: {e.message}") return # 5. Search with error handling try: response = collection.search_knowledge( SearchKnowledgeRequest( query="Q1 revenue", limit=5, ) ) print(f"Found {len(response.result.result_list)} results") except VikingKnowledgeException as e: print(f"Search failed: {e.code} - {e.message}") safe_knowledge_workflow() ``` -------------------------------- ### Handle Data Not Found Source: https://github.com/volcengine/vikingdb-python-sdk/blob/main/_autodocs/api-reference/memory-exceptions.md Example of catching a DataNotFoundException when trying to update a profile for a user ID that does not exist. ```python from vikingdb.memory import DataNotFoundException try: collection.update_profile( user_id="nonexistent_user", profile_info={"name": "Alice"}, ) except DataNotFoundException as e: print("User profile not found") ``` -------------------------------- ### Perform Text Search with Multi-Modal Request Source: https://github.com/volcengine/vikingdb-python-sdk/blob/main/_autodocs/api-reference/vector-types.md Example of creating a SearchByMultiModalRequest for a text-based search and executing it. ```python from vikingdb.vector import SearchByMultiModalRequest # Text search request = SearchByMultiModalRequest( text="beautiful landscape", limit=10, ) response = index.search_by_multi_modal(request) ``` -------------------------------- ### Handle Collection Not Found Source: https://github.com/volcengine/vikingdb-python-sdk/blob/main/_autodocs/api-reference/memory-exceptions.md Example of catching a CollectionNotExistException when trying to access or operate on a collection that does not exist. ```python from vikingdb.memory import CollectionNotExistException try: collection = client.get_collection( collection_name="nonexistent_collection", ) # Operations on collection fail here collection.add_event(...) except CollectionNotExistException as e: print("Collection not found") print(f"Verify collection name: {e.message}") ``` -------------------------------- ### Perform Random Sampling Source: https://github.com/volcengine/vikingdb-python-sdk/blob/main/_autodocs/api-reference/vector-types.md Example of creating a SearchByRandomRequest to retrieve a random sample of items with specified output fields. ```python from vikingdb.vector import SearchByRandomRequest request = SearchByRandomRequest( limit=5, output_fields=["id", "name"], ) response = index.search_by_random(request) ``` -------------------------------- ### Get Collection Client using Collection and Project Names Source: https://github.com/volcengine/vikingdb-python-sdk/blob/main/_autodocs/api-reference/vikingdb-vector-client.md Obtain a CollectionClient by specifying the collection name and project name. Use this when the resource ID is not readily available. ```python # Option 2: Using collection name collection = client.collection( collection_name="my_collection", project_name="my_project", ) ``` -------------------------------- ### Perform Search By ID Source: https://github.com/volcengine/vikingdb-python-sdk/blob/main/_autodocs/api-reference/vector-types.md Example of creating a SearchByIDRequest to find similar items and specifying which fields to output. ```python from vikingdb.vector import SearchByIDRequest request = SearchByIDRequest( id="product-123", limit=10, output_fields=["name", "price"], ) response = index.search_by_id(request) ``` -------------------------------- ### Perform Aggregation Operations Source: https://github.com/volcengine/vikingdb-python-sdk/blob/main/_autodocs/api-reference/vector-types.md Examples of creating AggRequests for counting all items and calculating the average price within a specific category. ```python from vikingdb.vector import AggRequest # Count all request = AggRequest(op="count") response = index.aggregate(request) # Average price request = AggRequest( op="avg", field="price", filter={"category": "electronics"}, ) response = index.aggregate(request) ``` -------------------------------- ### Perform Keyword Search Source: https://github.com/volcengine/vikingdb-python-sdk/blob/main/_autodocs/api-reference/vector-types.md Example of creating a SearchByKeywordsRequest for a natural language query, specifying fields and BM25 parameters. ```python from vikingdb.vector import SearchByKeywordsRequest # Natural language query request = SearchByKeywordsRequest( query="python programming tutorial", limit=10, fields=["title", "content"], bm25_k1=2.0, bm25_b=0.75, ) response = index.search_by_keywords(request) ``` -------------------------------- ### Rerank with Custom Instruction Source: https://github.com/volcengine/vikingdb-python-sdk/blob/main/_autodocs/api-reference/rerank-client.md Demonstrates how to use the rerank client with a custom instruction to guide the reranking process. This is useful for prioritizing specific aspects of the search results. ```APIDOC ## Rerank with Custom Instruction ### Description This example shows how to use the `rerank` method with a custom `instruction` to influence the reranking logic. The instruction helps to prioritize results based on specific criteria, such as community support. ### Method ```python rerank_client.rerank( RerankRequest( query="latest Python web frameworks", model="bge-reranker-large", instruction="Prioritize frameworks with strong community support and active development", items=[ Rerank(index=0, text="FastAPI - modern, fast web framework"), Rerank(index=1, text="Django - batteries-included framework"), Rerank(index=2, text="Flask - lightweight, flexible framework"), ], ) ) ``` ### Parameters #### Request Body - **query** (string) - Required - The search query. - **model** (string) - Required - The reranker model to use. - **instruction** (string) - Optional - A custom instruction to guide the reranking. - **items** (list[Rerank]) - Required - A list of items to rerank, each with an index and text. ``` -------------------------------- ### Perform Scalar Field Search Source: https://github.com/volcengine/vikingdb-python-sdk/blob/main/_autodocs/api-reference/vector-types.md Example of creating a SearchByScalarRequest to find the top 10 most expensive items. ```python from vikingdb.vector import SearchByScalarRequest # Top 10 most expensive request = SearchByScalarRequest( field="price", order="desc", limit=10, ) response = index.search_by_scalar(request) ``` -------------------------------- ### Complete Search + Rerank Workflow Source: https://github.com/volcengine/vikingdb-python-sdk/blob/main/_autodocs/api-reference/rerank-client.md A comprehensive example showing the integration of embedding, vector search, and reranking to achieve a complete search workflow. ```APIDOC ## Complete Search + Rerank Workflow ### Description This example outlines a full-stack search process: first, embedding a query, then performing a vector search to retrieve candidate documents, and finally reranking these candidates to present the most relevant results. ### Steps 1. **Embed query**: Convert the search query into a vector representation. 2. **Vector search (recall)**: Use the query vector to find similar items in an index. 3. **Rerank top candidates**: Reorder the retrieved items based on their relevance to the original query using a reranker model. 4. **Return final reranked results**: Display the top reranked items. ### Method ```python # Initialize VikingDB client auth = IAM(ak="ak", sk="sk") client = VikingDB( host="api-vikingdb.vikingdb.ap-southeast-1.volces.com", region="ap-southeast-1", auth=auth, ) # 1. Embed query query = "best machine learning frameworks" embedding_response = client.embedding().embedding( EmbeddingRequest( model="bge-large", texts=[query], ) ) query_vector = embedding_response.result.data[0].embedding # 2. Vector search (recall) index = client.index( collection_name="ml_library", index_name="lib_index", ) search_response = index.search_by_vector( SearchByVectorRequest( dense_vector=query_vector, limit=50, # Get many candidates output_fields=["name", "description"], ) ) # 3. Rerank top candidates rerank_items = [ Rerank( index=i, text=item.fields.get("description", ""), title=item.fields.get("name", ""), ) for i, item in enumerate(search_response.result.data[:20]) ] rerank_response = client.rerank().rerank( RerankRequest( query=query, model="bge-reranker-large", items=rerank_items, ) ) # 4. Return final reranked results print(f"Final results ({len(rerank_response.result.data)} items):") for item in rerank_response.result.data[:5]: print(item) ``` ``` -------------------------------- ### Rerank with Custom Instruction Source: https://github.com/volcengine/vikingdb-python-sdk/blob/main/_autodocs/api-reference/rerank-client.md Use this snippet when you need to guide the reranking process with specific criteria beyond just the query text. Ensure the 'instruction' parameter is clear and concise. ```python from vikingdb.vector import RerankRequest, Rerank response = rerank_client.rerank( RerankRequest( query="latest Python web frameworks", model="bge-reranker-large", instruction="Prioritize frameworks with strong community support and active development", items=[ Rerank(index=0, text="FastAPI - modern, fast web framework"), Rerank(index=1, text="Django - batteries-included framework"), Rerank(index=2, text="Flask - lightweight, flexible framework"), ], ) ) ``` -------------------------------- ### Get Collection by Name Source: https://github.com/volcengine/vikingdb-python-sdk/blob/main/_autodocs/api-reference/vikingmem-client.md Retrieve a Collection instance for performing memory operations by specifying the collection and project names. Defaults to the 'default' project if not provided. ```python # By name collection = client.get_collection( collection_name="demo_collection", project_name="default", ) ``` -------------------------------- ### Run All Scenario Tests with Pytest Source: https://github.com/volcengine/vikingdb-python-sdk/blob/main/examples/vector/README.md Executes all scenario tests within the vector examples directory using pytest. This is a comprehensive test run for the vector functionalities. ```bash env $(grep -v '^#' .env | xargs) pytest -q examples/vector -k scenario ``` -------------------------------- ### Get Collection Client using Resource ID Source: https://github.com/volcengine/vikingdb-python-sdk/blob/main/_autodocs/api-reference/vikingdb-vector-client.md Obtain a CollectionClient by providing a unique resource ID. This is the preferred method when the resource ID is available. ```python # Option 1: Using resource ID collection = client.collection(resource_id="resource-123") ``` -------------------------------- ### Exception Hierarchy Handling Source: https://github.com/volcengine/vikingdb-python-sdk/blob/main/_autodocs/api-reference/memory-exceptions.md Demonstrates handling a hierarchy of exceptions, starting with specific errors like UnauthorizedException, CollectionNotExistException, and then broader categories like search/fetch errors, before a final catch-all for VikingMemException. ```python try: # Memory operations pass except UnauthorizedException: # Handle auth error specifically pass except (CollectionNotExistException, DataNotFoundException): # Handle data not found errors pass except (IndexSearchException, IndexFetchException): # Handle search/fetch errors pass except VikingMemException as e: # Fallback for other memory errors print(f"Memory error {e.code}: {e.message}") ``` -------------------------------- ### Define Multi-Modal Search Request Source: https://github.com/volcengine/vikingdb-python-sdk/blob/main/_autodocs/api-reference/vector-types.md Use SearchByMultiModalRequest to initiate searches using text, image, or video inputs. Specify optional parameters like need_instruction for guided searches. ```python class SearchByMultiModalRequest(SearchBase): text: Optional[str] = None image: Optional[Any] = None video: Optional[Any] = None need_instruction: Optional[bool] = None instruction: Optional[Instruction] = None tensor_rerank: Optional[TensorRerank] = None rerank: Optional[ModelRerank] = None ``` -------------------------------- ### Example Usage of DeleteDataRequest Source: https://github.com/volcengine/vikingdb-python-sdk/blob/main/_autodocs/api-reference/vector-types.md Shows how to delete specific records by providing a list of IDs or how to delete all records in a collection by setting 'delete_all' to True. Ensure the 'ids' field is a sequence of identifiers for targeted deletion. ```python from vikingdb.vector import DeleteDataRequest # Delete specific IDs request = DeleteDataRequest(ids=["1", "2", "3"]) response = collection.delete(request) # Delete all request = DeleteDataRequest(delete_all=True) response = collection.delete(request) ``` -------------------------------- ### Example Usage of UpsertDataRequest Source: https://github.com/volcengine/vikingdb-python-sdk/blob/main/_autodocs/api-reference/vector-types.md Demonstrates how to create and use an UpsertDataRequest to insert multiple records with specified fields and a time-to-live. Ensure the 'data' field contains a sequence of mappings, where each mapping includes an 'id' and a 'fields' dictionary. ```python from vikingdb.vector import UpsertDataRequest request = UpsertDataRequest( data=[ {"id": "1", "fields": {"text": "hello", "score": 0.95}}, {"id": "2", "fields": {"text": "world", "score": 0.87}}, ], ttl=3600, ) response = collection.upsert(request) ``` -------------------------------- ### VikingMemException Usage Example Source: https://github.com/volcengine/vikingdb-python-sdk/blob/main/_autodocs/exceptions.md The base exception for memory operations. This example shows how to catch memory-related errors when accessing collections. ```python class VikingMemException(VikingException): pass ``` ```python from vikingdb.memory import VikingMem, VikingMemException try: collection = client.get_collection( collection_name="nonexistent", ) except VikingMemException as e: print(f"Memory error {e.code}: {e.message}") ``` -------------------------------- ### VikingVectorException Usage Example Source: https://github.com/volcengine/vikingdb-python-sdk/blob/main/_autodocs/exceptions.md Raised when vector operations fail. This example shows how to catch and handle vector-specific errors during a search operation. ```python class VikingVectorException(VikingException): def __init__( self, code: Union[int, str], request_id: str = "unknown", message: Optional[str] = None, status_code: Optional[int] = None, ) -> None ``` ```python from vikingdb.vector import VikingVectorException, SearchByVectorRequest try: response = index.search_by_vector( SearchByVectorRequest(dense_vector=[...], limit=10) ) except VikingVectorException as e: print(f"Vector search failed: {e.code} - {e.message}") ``` -------------------------------- ### Initialize VikingDB Client and IndexClient Source: https://github.com/volcengine/vikingdb-python-sdk/blob/main/_autodocs/api-reference/index-client.md Demonstrates how to initialize the VikingDB client and obtain an IndexClient instance. This is a prerequisite for performing index-scoped operations. ```python from vikingdb import VikingDB, IAM from vikingdb.vector import SearchByVectorRequest client = VikingDB( host="api-vikingdb.vikingdb.ap-southeast-1.volces.com", region="ap-southeast-1", auth=IAM(ak="ak", sk="sk"), ) index = client.index( collection_name="my_collection", index_name="my_index", ) ``` -------------------------------- ### Configure Multi-Region Clients with Failover Source: https://github.com/volcengine/vikingdb-python-sdk/blob/main/_autodocs/configuration.md Set up clients for a main and a failover region, with a function to handle failover during search operations. ```python from vikingdb import VikingDB, IAM # Main region client main_client = VikingDB( host="api-vikingdb.vikingdb.ap-southeast-1.volces.com", region="ap-southeast-1", auth=IAM(ak="ak", sk="sk"), ) # Failover region client failover_client = VikingDB( host="api-vikingdb.vikingdb.cn-beijing.volces.com", region="cn-beijing", auth=IAM(ak="ak", sk="sk"), ) def search_with_failover(query): try: return main_client.index(...).search_by_vector(query) except Exception: return failover_client.index(...).search_by_vector(query) ``` -------------------------------- ### Basic Usage with Vector Client Source: https://github.com/volcengine/vikingdb-python-sdk/blob/main/_autodocs/api-reference/request-options.md Demonstrates how to instantiate RequestOptions with custom settings and pass them to a client method like `search_by_vector`. ```APIDOC ## Usage ### Basic Usage with Vector Client ```python from vikingdb import IAM, VikingDB, RequestOptions from vikingdb.vector import SearchByVectorRequest auth = IAM(ak="your-ak", sk="your-sk") client = VikingDB( host="api-vikingdb.vikingdb.ap-southeast-1.volces.com", region="ap-southeast-1", auth=auth, ) # Apply per-request options options = RequestOptions( request_id="custom-request-123", timeout=10, max_attempts=5, ) index = client.index( collection_name="my_collection", index_name="my_index", ) response = index.search_by_vector( SearchByVectorRequest( dense_vector=[0.1, 0.2, 0.3], limit=10, ), request_options=options, ) ``` ``` -------------------------------- ### Initialize Clients with Environment Variables Source: https://github.com/volcengine/vikingdb-python-sdk/blob/main/_autodocs/configuration.md Load configuration for VikingDB and related services from environment variables during client initialization. ```python import os from vikingdb import IAM, VikingDB, APIKey from vikingdb.vector import SearchByVectorRequest # Vector Database Configuration VIKINGDB_HOST = os.getenv("VIKINGDB_HOST", "api-vikingdb.vikingdb.ap-southeast-1.volces.com") VIKINGDB_REGION = os.getenv("VIKINGDB_REGION", "ap-southeast-1") VIKINGDB_AK = os.getenv("VIKINGDB_AK") VIKINGDB_SK = os.getenv("VIKINGDB_SK") VIKINGDB_COLLECTION = os.getenv("VIKINGDB_COLLECTION", "demo_collection") VIKINGDB_INDEX = os.getenv("VIKINGDB_INDEX", "demo_index") # Memory Configuration VIKINGDB_MEMORY_HOST = os.getenv("VIKINGDB_MEMORY_HOST", "api-knowledgebase.mlp.cn-beijing.volces.com") VIKINGDB_MEMORY_REGION = os.getenv("VIKINGDB_MEMORY_REGION", "cn-beijing") # Knowledge Configuration VIKING_COLLECTION_RID = os.getenv("VIKING_COLLECTION_RID") VIKING_COLLECTION_NAME = os.getenv("VIKING_COLLECTION_NAME") VIKING_PROJECT = os.getenv("VIKING_PROJECT", "default") VIKING_CHAT_API_KEY = os.getenv("VIKING_CHAT_API_KEY") # Initialize clients auth = IAM(ak=VIKINGDB_AK, sk=VIKINGDB_SK) vector_client = VikingDB( host=VIKINGDB_HOST, region=VIKINGDB_REGION, auth=auth, timeout=30, ) knowledge_client = VikingKnowledge( host=VIKINGDB_MEMORY_HOST, region=VIKINGDB_MEMORY_REGION, auth=APIKey(api_key=VIKING_CHAT_API_KEY), ) ``` -------------------------------- ### Initialize VikingMem Memory Client Source: https://github.com/volcengine/vikingdb-python-sdk/blob/main/_autodocs/configuration.md Instantiate the VikingMem Memory client. Default host and region are provided, but can be overridden. Authentication is required. ```python from vikingdb import VikingMem, IAM client = VikingMem( host="api-knowledgebase.mlp.cn-beijing.volces.com", region="cn-beijing", auth=IAM(ak="ak", sk="sk"), scheme="http", timeout=30, ) ``` -------------------------------- ### Initialize VikingKnowledge Client Source: https://github.com/volcengine/vikingdb-python-sdk/blob/main/_autodocs/configuration.md Instantiate the VikingKnowledge client, which shares parameters with VikingMem. Authentication is required. ```python from vikingdb import VikingKnowledge, APIKey client = VikingKnowledge( host="api-knowledgebase.mlp.cn-beijing.volces.com", region="cn-beijing", auth=APIKey(api_key="your-api-key"), ) ``` -------------------------------- ### Initialize VikingDB Client Source: https://github.com/volcengine/vikingdb-python-sdk/blob/main/_autodocs/api-reference/vikingdb-vector-client.md Instantiate the VikingDB client with host, region, and authentication details. Supports optional scheme, STS token, and timeout configurations. ```python from vikingdb import VikingDB, IAM client = VikingDB( host="api-vikingdb.vikingdb.ap-southeast-1.volces.com", region="ap-southeast-1", auth=IAM(ak="ak", sk="sk"), ) ``` -------------------------------- ### Configure Development Environment Client Source: https://github.com/volcengine/vikingdb-python-sdk/blob/main/_autodocs/configuration.md Set up a VikingDB client for a development environment, using HTTP and a longer timeout. ```python from vikingdb import VikingDB, IAM # Development with HTTP and longer timeout dev_client = VikingDB( host="localhost:8080", region="local", auth=IAM(ak="dev-ak", sk="dev-sk"), scheme="http", timeout=60, ) ``` -------------------------------- ### Initialize VikingDB Vector Client Source: https://github.com/volcengine/vikingdb-python-sdk/blob/main/_autodocs/configuration.md Instantiate the VikingDB Vector client with host, region, and authentication details. The scheme and timeout can also be configured. ```python from vikingdb import VikingDB, IAM client = VikingDB( host="api-vikingdb.vikingdb.ap-southeast-1.volces.com", region="ap-southeast-1", auth=IAM(ak="your-ak", sk="your-sk"), scheme="https", timeout=30, ) ``` -------------------------------- ### Initialize VikingDB Client Source: https://github.com/volcengine/vikingdb-python-sdk/blob/main/_autodocs/api-reference/vikingdb-vector-client.md Initialize the VikingDB client with authentication credentials and connection details. Ensure environment variables for AK, SK, HOST, and REGION are set. ```python import os from vikingdb import IAM, VikingDB from vikingdb.vector import SearchByRandomRequest # Initialize client auth = IAM( ak=os.environ["VIKINGDB_AK"], sk=os.environ["VIKINGDB_SK"], ) client = VikingDB( host=os.environ["VIKINGDB_HOST"], region=os.environ["VIKINGDB_REGION"], auth=auth, scheme="https", timeout=30, ) # Use collection collection = client.collection( collection_name="demo_collection", ) # Use index index = client.index( collection_name="demo_collection", index_name="demo_index", ) response = index.search_by_random( SearchByRandomRequest(limit=5) ) print(f"Found {len(response.result.data or [])} results") ``` -------------------------------- ### Add Events and Sessions, Search Memories Source: https://github.com/volcengine/vikingdb-python-sdk/blob/main/_autodocs/api-reference/vikingmem-client.md Demonstrates how to add purchase and login events, create a user session with messages, and search for memories with a filter. Use this for basic data ingestion and retrieval. ```python event1 = collection.add_event( event_type="purchase", memory_info={ "product": "laptop", "price": 1000, }, user_id="user_001", ) event2 = collection.add_event( event_type="login", memory_info={ "timestamp": 1234567890, "device": "mobile", }, user_id="user_001", ) # 3. Add session session = collection.add_session( session_id="session_001", messages=[ {"role": "user", "content": "Recommend a laptop"}, {"role": "assistant", "content": "Based on your budget..."}, ], metadata={ "user_id": "user_001", "assistant_id": "bot_v1", } ) # 4. Search memories results = collection.search_memory( query="products purchased", filter={"user_id": "user_001"}, limit=5, ) print(f"Found {len(results.get('items', []))} results") ``` -------------------------------- ### Initialize Knowledge Base Client with IAM Auth Source: https://github.com/volcengine/vikingdb-python-sdk/blob/main/README.md Initialize the VikingKnowledge client for knowledge base operations using IAM authentication. Ensure environment variables for resource ID, collection name, and project are set. ```python import os from vikingdb import IAM, APIKey from vikingdb.knowledge import ( VikingKnowledge, KnowledgeCollection, SearchKnowledgeRequest, ) # IAM auth for collection-level operations client = VikingKnowledge( auth=IAM(ak=os.getenv("VOLC_AK"), sk=os.getenv("VOLC_SK")), host="api-knowledgebase.mlp.cn-beijing.volces.com", region="cn-beijing", scheme="http", ) collection = client.collection( resource_id=os.getenv("VIKING_COLLECTION_RID"), collection_name=os.getenv("VIKING_COLLECTION_NAME") or "financial_reports", project_name=os.getenv("VIKING_PROJECT") or "default", ) resp = collection.search_knowledge( SearchKnowledgeRequest(query="2025 Q1 revenue growth", limit=5, dense_weight=0.5) ) print(f"request_id={resp.request_id} hits={len(resp.result.result_list or [])}") # API-key auth for service-level chat svc_client = VikingKnowledge(auth=APIKey(api_key=os.getenv("VIKING_SERVICE_API_KEY"))) ``` -------------------------------- ### Configure Production Environment Client Source: https://github.com/volcengine/vikingdb-python-sdk/blob/main/_autodocs/configuration.md Set up a VikingDB client for a production environment, using HTTPS and a standard timeout, with credentials loaded from environment variables. ```python from vikingdb import VikingDB, IAM import os # Production with HTTPS and standard timeout prod_client = VikingDB( host=os.getenv("PROD_VIKINGDB_HOST"), region=os.getenv("PROD_VIKINGDB_REGION"), auth=IAM( ak=os.getenv("PROD_VIKINGDB_AK"), sk=os.getenv("PROD_VIKINGDB_SK"), ), scheme="https", timeout=30, ) ``` -------------------------------- ### Initialize VikingMem Client Source: https://github.com/volcengine/vikingdb-python-sdk/blob/main/_autodocs/api-reference/vikingmem-client.md Instantiate the VikingMem client with authentication credentials and connection details. Requires an IAM or APIKey authentication provider. ```python from vikingdb import IAM from vikingdb.memory import VikingMem auth = IAM(ak="ak", sk="sk") client = VikingMem( host="api-knowledgebase.mlp.cn-beijing.volces.com", region="cn-beijing", auth=auth, ) ``` -------------------------------- ### Retry Configuration Source: https://github.com/volcengine/vikingdb-python-sdk/blob/main/_autodocs/api-reference/request-options.md Example of overriding the default retry attempts for a specific request by setting the `max_attempts` attribute in `RequestOptions`. ```APIDOC ### Retry Configuration ```python # Override retry attempts for a specific request options = RequestOptions(max_attempts=10) # Attempts will retry up to 10 times with exponential backoff (0.5s → 8s max) response = index.search_by_vector(request, request_options=options) ``` ``` -------------------------------- ### Custom Headers Source: https://github.com/volcengine/vikingdb-python-sdk/blob/main/_autodocs/api-reference/request-options.md Example of setting custom HTTP headers for a specific request using the `headers` attribute of `RequestOptions`. ```APIDOC ### Custom Headers ```python options = RequestOptions( headers={ "X-Custom-Header": "custom-value", "X-Request-Priority": "high", } ) response = index.search_by_vector(request, request_options=options) ``` ``` -------------------------------- ### VikingConnectionException Handling Source: https://github.com/volcengine/vikingdb-python-sdk/blob/main/_autodocs/exceptions.md Raised when unable to connect to the vector service. This example demonstrates catching connection errors during client initialization. ```python class VikingConnectionException(Exception): def __init__(self, msg: str, cause: str) -> None ``` ```python from vikingdb import VikingDB, IAM from vikingdb.vector import VikingConnectionException try: client = VikingDB( host="unreachable.example.com", region="ap-southeast-1", auth=IAM(ak="ak", sk="sk"), ) except VikingConnectionException as e: print(f"Failed to connect: {e}") ``` -------------------------------- ### Initialize VikingDB Client Source: https://github.com/volcengine/vikingdb-python-sdk/blob/main/_autodocs/README.md Initializes the VikingDB client with IAM authentication. Ensure your AK and SK are correctly set. ```python from vikingdb import VikingDB, IAM auth = IAM(ak="your-ak", sk="your-sk") client = VikingDB( host="api-vikingdb.vikingdb.ap-southeast-1.volces.com", region="ap-southeast-1", auth=auth, ) index = client.index( collection_name="my_collection", index_name="my_index", ) ``` -------------------------------- ### Handle Upsert Operation Failure Source: https://github.com/volcengine/vikingdb-python-sdk/blob/main/_autodocs/api-reference/memory-exceptions.md Example of catching an UpsertOpFailedException when attempting to add an event with invalid data, such as a negative amount. ```python try: collection.add_event( event_type="purchase", memory_info={ "amount": -100, # Invalid: negative amount }, user_id="user_001", ) except UpsertOpFailedException as e: print(f"Failed to add event: {e.message}") ``` -------------------------------- ### Handle Invalid Request Source: https://github.com/volcengine/vikingdb-python-sdk/blob/main/_autodocs/api-reference/memory-exceptions.md Example of catching an InvalidRequestException when submitting a request with invalid parameters, like an empty event type. ```python try: collection.add_event( event_type="", # Empty event type memory_info={}, ) except InvalidRequestException as e: print(f"Invalid request: {e.message}") ``` -------------------------------- ### Get KnowledgeCollection Instance Source: https://github.com/volcengine/vikingdb-python-sdk/blob/main/_autodocs/api-reference/vikingknowledge-client.md Obtains a KnowledgeCollection instance for performing document and point operations. Can be identified by resource ID or by name and project. ```python def collection( self, *, resource_id: Optional[str] = None, collection_name: Optional[str] = None, project_name: Optional[str] = None, ) -> KnowledgeCollection: ``` ```python from vikingdb import VikingKnowledge, IAM client = VikingKnowledge( host="api-knowledgebase.mlp.cn-beijing.volces.com", region="cn-beijing", auth=IAM(ak="ak", sk="sk"), ) collection = client.collection( resource_id="collection-rid-123", ) # Or by name collection = client.collection( collection_name="financial_reports", project_name="default", ) ```