Examples

Python Examples

Copy page

Complete Python examples for interacting with the Cortex API, including a reusable client class and common operations.

Setup

Install the required library:

Code
 
pip install requests

Cortex Client Class

A reusable client wrapper for all API operations:

Code
 
"""Cortex Python Client"""

import requests
from typing import Optional, Dict, Any, List, Iterator
from pathlib import Path


class CortexClient:
    """Client for the Cortex API."""
    
    def __init__(self, base_url: str, api_key: str):
        """
        Initialize the Cortex client.
        
        Args:
            base_url: The API base URL (e.g., "http://localhost:8000")
            api_key: Your Cortex API key
        """
        self.base_url = base_url.rstrip('/')
        self.session = requests.Session()
        self.session.headers.update({
            'X-API-Key': api_key,
            'Content-Type': 'application/json'
        })
    
    def _request(
        self, 
        method: str, 
        path: str, 
        **kwargs
    ) -> Dict[str, Any]:
        """Make an API request."""
        url = f"{self.base_url}{path}"
        response = self.session.request(method, url, **kwargs)
        response.raise_for_status()
        return response.json()
    
    # =========================================================================
    # Health & Stats
    # =========================================================================
    
    def health(self) -> Dict[str, Any]:
        """Check API health."""
        return self._request('GET', '/health')
    
    def stats(self) -> Dict[str, Any]:
        """Get knowledge base statistics."""
        return self._request('GET', '/api/stats')
    
    # =========================================================================
    # Documents
    # =========================================================================
    
    def upload_document(
        self, 
        file_path: str, 
        collection_id: Optional[str] = None,
        start_processing: bool = True
    ) -> Dict[str, Any]:
        """
        Upload a document to the knowledge base.
        
        Args:
            file_path: Path to the file to upload
            collection_id: Optional collection to add the document to
            start_processing: Whether to start processing immediately
        """
        path = Path(file_path)
        with open(path, 'rb') as f:
            files = {'file': (path.name, f, 'application/octet-stream')}
            # collection_id and start_processing are query parameters, not form data
            params = {'start_processing': str(start_processing).lower()}
            if collection_id:
                params['collection_id'] = collection_id
            
            # Remove Content-Type for multipart
            headers = {'X-API-Key': self.session.headers['X-API-Key']}
            response = requests.post(
                f"{self.base_url}/api/upload",
                files=files,
                params=params,  # Use params for query parameters
                headers=headers
            )
            response.raise_for_status()
            return response.json()
    
    def list_documents(
        self, 
        collection_id: Optional[str] = None,
        status: Optional[str] = None,
        limit: int = 100
    ) -> List[Dict[str, Any]]:
        """List documents in the knowledge base."""
        params = {'limit': limit}
        if collection_id:
            params['collection_id'] = collection_id
        if status:
            params['status'] = status
        return self._request('GET', '/api/documents', params=params)
    
    def get_document(self, doc_id: str) -> Dict[str, Any]:
        """Get document details."""
        return self._request('GET', f'/api/documents/{doc_id}')
    
    def delete_document(self, doc_id: str) -> Dict[str, Any]:
        """Delete a document."""
        return self._request('DELETE', f'/api/documents/{doc_id}')
    
    def reprocess_document(self, doc_id: str) -> Dict[str, Any]:
        """Reprocess a document."""
        return self._request('POST', f'/api/documents/{doc_id}/reprocess')
    
    # =========================================================================
    # Search
    # =========================================================================
    
    def search(
        self, 
        query: str, 
        limit: int = 10,
        collection_id: Optional[str] = None,
        search_type: str = "hybrid"
    ) -> Dict[str, Any]:
        """
        Search the knowledge base.
        
        Args:
            query: Search query
            limit: Maximum results to return
            collection_id: Optional collection to search within
            search_type: "hybrid", "vector", "keyword", or "graph"
        """
        payload = {
            'query': query,
            'limit': limit,
            'search_type': search_type
        }
        if collection_id:
            payload['collection_id'] = collection_id
        return self._request('POST', '/api/search', json=payload)
    
    # =========================================================================
    # Ask AI (RAG)
    # =========================================================================
    
    def ask(
        self,
        question: str,
        collection_id: Optional[str] = None,
        conversation_history: Optional[List[Dict]] = None,
        use_fast_search: bool = False,
        use_agentic: bool = False
    ) -> Dict[str, Any]:
        """
        Ask a question using RAG.

        Args:
            question: The question to ask
            collection_id: Optional collection to query
            conversation_history: Previous messages for context
            use_fast_search: Use fast vector-only search
            use_agentic: Use deep research mode
        """
        payload = {'question': question}
        if collection_id:
            payload['collection_id'] = collection_id
        if conversation_history:
            payload['conversation_history'] = conversation_history
        if use_fast_search:
            payload['use_fast_search'] = True
        if use_agentic:
            payload['use_agentic'] = True
        return self._request('POST', '/api/ask', json=payload)

    def ask_streaming(
        self,
        question: str,
        collection_id: Optional[str] = None,
        use_agentic: bool = False,
        use_fast_search: bool = False
    ) -> Iterator[Dict[str, Any]]:
        """
        Ask a question with streaming SSE response.

        Yields parsed event dicts as they arrive.
        """
        import json
        payload = {'question': question}
        if collection_id:
            payload['collection_id'] = collection_id
        if use_agentic:
            payload['use_agentic'] = True
        if use_fast_search:
            payload['use_fast_search'] = True

        response = self.session.post(
            f"{self.base_url}/api/ask/stream",
            json=payload,
            stream=True
        )
        response.raise_for_status()

        for line in response.iter_lines(decode_unicode=True):
            if line and line.startswith("data: "):
                yield json.loads(line[6:])
    
    # =========================================================================
    # Collections
    # =========================================================================
    
    def create_collection(
        self, 
        name: str, 
        description: Optional[str] = None
    ) -> Dict[str, Any]:
        """Create a new collection."""
        payload = {'name': name}
        if description:
            payload['description'] = description
        return self._request('POST', '/api/collections', json=payload)
    
    def list_collections(self) -> List[Dict[str, Any]]:
        """List all collections."""
        return self._request('GET', '/api/collections')
    
    def delete_collection(self, collection_id: str) -> Dict[str, Any]:
        """Delete a collection."""
        return self._request('DELETE', f'/api/collections/{collection_id}')
    
    # =========================================================================
    # Knowledge Graph
    # =========================================================================
    
    def get_graph_visualization(
        self, 
        collection_id: Optional[str] = None
    ) -> Dict[str, Any]:
        """Get graph visualization data."""
        params = {}
        if collection_id:
            params['collection_id'] = collection_id
        return self._request('GET', '/api/graph/visualization', params=params)
    
    def search_entities(
        self,
        search: str,
        entity_type: Optional[str] = None,
        limit: int = 20
    ) -> List[Dict[str, Any]]:
        """Search for entities in the knowledge graph."""
        params = {'search': search, 'limit': limit}
        if entity_type:
            params['type'] = entity_type
        return self._request('GET', '/api/graph/entities', params=params)

    def find_duplicate_entities(
        self,
        threshold: float = 0.85,
        limit: int = 50
    ) -> Dict[str, Any]:
        """
        Find duplicate entity candidates.

        Args:
            threshold: Similarity threshold (0.5 to 1.0)
            limit: Maximum number of duplicate groups to return
        """
        params = {'threshold': threshold, 'limit': limit}
        return self._request('GET', '/api/entities/duplicates', params=params)

    def merge_entities(
        self,
        canonical: str,
        merge: List[str]
    ) -> Dict[str, Any]:
        """
        Merge duplicate entities into a canonical entity.

        Args:
            canonical: Name of the entity to keep
            merge: List of entity names to merge into canonical
        """
        payload = {'canonical': canonical, 'merge': merge}
        return self._request('POST', '/api/entities/merge', json=payload)

    def get_merge_history(self, limit: int = 50) -> Dict[str, Any]:
        """Get entity merge history."""
        return self._request('GET', '/api/entities/merge-history', params={'limit': limit})

Usage Examples

Basic Setup

Code
 
# Initialize the client
client = CortexClient(
    base_url="http://localhost:8000",
    api_key="your-api-key"
)

# Check health
health = client.health()
print(f"Status: {health['status']}")

Upload Documents

Code
 
# Upload a single document
result = client.upload_document(
    "research_paper.pdf",
    collection_id="research"
)
print(f"Uploaded: {result['doc_id']}")

# Upload multiple documents
import glob

for file_path in glob.glob("documents/*.pdf"):
    result = client.upload_document(file_path)
    print(f"Uploaded: {result['filename']}")

Search

Code
 
# Basic search
results = client.search("machine learning")
for r in results['results']:
    print(f"- {r['document_title']}: {r['content'][:100]}...")

# Search in a specific collection
results = client.search(
    query="neural networks",
    collection_id="research",
    limit=5
)

Ask Questions

Code
 
# Simple question
response = client.ask("What are the main findings?")
print(response['answer'])

# Print sources
for source in response.get('sources', []):
    print(f"  Source: {source['document_title']}")

# Multi-turn conversation
history = []
question = "What is GraphRAG?"
response = client.ask(question, conversation_history=history)
history.append({"role": "user", "content": question})
history.append({"role": "assistant", "content": response['answer']})

# Follow-up question
response = client.ask(
    "How does it compare to traditional RAG?",
    conversation_history=history
)

Streaming Responses

Code
 
# Stream the response
for chunk in client.ask_streaming("Explain knowledge graphs"):
    print(chunk, end='', flush=True)
print()

Entity Deduplication

Code
 
# Find potential duplicate entities
duplicates = client.find_duplicate_entities(threshold=0.85)
print(f"Found {duplicates['total_groups']} duplicate groups")

for group in duplicates['groups']:
    canonical = group['canonical']
    dupes = [d['name'] for d in group['duplicates']]
    print(f"  {canonical} <- {dupes}")

# Merge confirmed duplicates
result = client.merge_entities(
    canonical="Machine Learning",
    merge=["machine learning", "ML"]
)
print(f"Merged {len(result['merged'])} entities")
print(f"  Relationships transferred: {result['relationships_transferred']}")
print(f"  Mentions transferred: {result['mentions_transferred']}")

# Review merge history
history = client.get_merge_history(limit=10)
for entry in history['entries']:
    print(f"  {entry['merged_at']}: {entry['merged']} -> {entry['canonical']}")

Error Handling

Code
 
from requests.exceptions import HTTPError

try:
    result = client.search("query")
except HTTPError as e:
    if e.response.status_code == 401:
        print("Invalid API key")
    elif e.response.status_code == 429:
        print("Rate limited, please wait")
    elif e.response.status_code == 403:
        print("Insufficient permissions")
    else:
        print(f"Error: {e}")

Complete Example Script

Code
 
#!/usr/bin/env python3
"""Example: Upload documents and ask questions."""

from cortex_client import CortexClient
import sys

def main():
    # Initialize client
    client = CortexClient(
        base_url="http://localhost:8000",
        api_key="your-api-key"
    )
    
    # Check health
    health = client.health()
    if health['status'] != 'healthy':
        print("API not healthy!")
        sys.exit(1)
    
    # Get stats
    stats = client.stats()
    print(f"Documents: {stats['document_count']}")
    print(f"Entities: {stats['entity_count']}")
    
    # Search
    results = client.search("important topic", limit=5)
    print(f"\nFound {len(results['results'])} results")
    
    # Ask a question
    response = client.ask("Summarize the key points")
    print(f"\nAnswer: {response['answer']}")

if __name__ == "__main__":
    main()