Problem: API key environment variable not being read.Solutions:
Verify environment variable name is exactly KUBIYA_API_KEY
Restart your terminal/IDE after setting the variable
Use .env file with python-dotenv:
from dotenv import load_dotenvload_dotenv() # Load from .env filefrom kubiya import ControlPlaneClientclient = ControlPlaneClient() # Will use KUBIYA_API_KEY
import timewhile True: status = client.datasets.get_dataset_status(dataset_id="dataset-id") if status['status'] == 'ready': break time.sleep(5)
Broaden your search query:
# Too specific - may return nothingresult = client.graph.semantic_search( query="exact phrase that might not exist", limit=10)# Broader queryresult = client.graph.semantic_search( query="general concept or keywords", limit=10)
Problem: Semantic search returns results with low similarity scores.Solutions:
Filter results by minimum score:
results = client.graph.semantic_search(query="your query", limit=50)# Only use high-confidence resultsrelevant = [r for r in results if r['similarity_score'] > 0.7]
Use intelligent search instead for complex queries:
# Better for complex questionsresult = client.graph.intelligent_search( keywords="complex multi-part question")
result = client.ingestion.ingest_nodes_batch( nodes=nodes, transactional=False # Don't rollback all on single failure)# Check resultsprint(f"Success: {result['summary']['success']}")print(f"Failed: {result['summary']['failed']}")if result.get('errors'): for error in result['errors']: print(f"Error: {error}")
Reduce batch size:
def import_in_chunks(nodes, chunk_size=100): """Import large datasets in smaller chunks.""" for i in range(0, len(nodes), chunk_size): chunk = nodes[i:i + chunk_size] result = client.ingestion.ingest_nodes_batch(nodes=chunk) print(f"Chunk {i//chunk_size + 1}: {result['summary']['success']} succeeded")
Problem: Application using too much memory.Solutions:
Process data in chunks:
# Instead of loading all at onceskip = 0while True: batch = client.graph.list_memories(skip=skip, limit=100) if not batch: break process_batch(batch) # Process and discard skip += 100
Clear sensitive data from memory:
import gcsecret = client.secrets.get_value(name="api-token")# Use secrettoken = secret['value']# ... use token ...# Clear from memorytoken = Nonesecret = Nonegc.collect()