API Reference

@vectorize Parameters

Usage

@vectorize(
    semantic_cache=True,
    cache_threshold=0.95,
    capture_return_value=True,
    capture_inputs=True,
    replay=True,
    auto=True,
    team="ml-team",  # custom tag (defined in .weaviate_properties)
)
async def generate_response(query: str):
    return await llm.complete(query)

@trace_span

Decorator for child span tracing. Must be used within a @vectorize parent context.

from vectorwave import trace_span

@trace_span
def child_function(data):
    return process(data)

Automatically inherits trace_id from the parent @vectorize call.

Optional Parameters

@trace_span also accepts keyword arguments for fine-grained control:

@trace_span(capture_return_value=True, attributes_to_capture=["score"])
def rank_results(results, score=0.5):
    return sorted(results, key=lambda x: x["score"])

Environment Variables

Connection

Vectorizer

GitHub (Self-Healing)

Drift Detection and Alerting

Performance Tuning

File Paths

Data and Tags

Core Functions

initialize_database()

Creates four Weaviate collections (VectorWaveFunctions, VectorWaveExecutions, VectorWaveGoldenDataset, VectorWaveTokenUsage). Must be called once before first use.

from vectorwave import initialize_database
initialize_database()

update_database_schema()

Performs zero-downtime schema migration when upgrading VectorWave.

from vectorwave import update_database_schema
update_database_schema()

search_functions(query, limit)

Vector search over stored function metadata.

from vectorwave import search_functions

results = search_functions(query="authentication", limit=5)

search_executions(limit, filters, sort_by, sort_ascending)

Search execution logs with filtering and sorting. Unlike search_functions, this uses filter-based lookup rather than vector search.

from vectorwave import search_executions

# Recent errors
results = search_executions(
    limit=10,
    filters={"status": "ERROR", "duration_ms__gte": 100},
    sort_by="timestamp_utc",
    sort_ascending=False,
)

Filter operators: exact match (default), __not_equal, __gte, __gt, __lte, __lt, __like.

search_functions_hybrid(query, limit, filters, alpha)

Hybrid search (keyword + vector) over function definitions.

from vectorwave import search_functions_hybrid

results = search_functions_hybrid(
    query="calculate loyalty points",
    limit=5,
    alpha=0.5,  # 0.0 = pure keyword, 1.0 = pure vector
)

search_errors_by_message(query, limit, filters)

Semantic search over error messages. Uses vector similarity to find errors matching a natural language description.

from vectorwave import search_errors_by_message

results = search_errors_by_message(
    query="timeout connecting to database",
    limit=10,
    filters={"function_name": "generate_response"},
)

generate_and_register_metadata()

Triggers LLM-based metadata generation for all registered functions that have auto=True. Useful for batch-generating descriptions after initial setup.

from vectorwave import generate_and_register_metadata

generate_and_register_metadata()

search_and_answer(query)

RAG-powered Q&A over your codebase.

from vectorwave import search_and_answer

answer = search_and_answer("How does the auth flow work?")

analyze_trace_log(trace_id, question)

Analyze a specific trace with a natural language question.

from vectorwave import analyze_trace_log

analysis = analyze_trace_log(
    trace_id="abc-123",
    question="Why was this trace slow?",
)

VectorWaveHealer

Constructor

from vectorwave import VectorWaveHealer

healer = VectorWaveHealer(
    model="gpt-4-turbo",     # LLM model (only parameter)
)

diagnose_and_heal()

result = healer.diagnose_and_heal(
    function_name="generate_response",
    lookback_minutes=60,      # Search errors in last N minutes
    create_pr=True,           # Create GitHub PR with fix
)
# Returns: str (LLM-suggested fix code or error message)

VectorWaveReplayer

Constructor

from vectorwave import VectorWaveReplayer
replayer = VectorWaveReplayer()

replay()

results = replayer.replay(
    function_full_name="app.generate_response",
    limit=20,                 # Max executions to replay
    update_baseline=False,    # Update baseline values on mismatch
)
# Returns: { function, total, passed, failed, updated, failures }

For semantic comparison, use SemanticReplayer:

from vectorwave import SemanticReplayer

semantic_replayer = SemanticReplayer()
results = semantic_replayer.replay(
    function_full_name="app.generate_response",
    limit=20,
    similarity_threshold=0.85,  # Vector similarity threshold
    semantic_eval=True,          # Use LLM-based semantic evaluation
)

VectorWaveDatasetManager

Constructor

from vectorwave import VectorWaveDatasetManager
dm = VectorWaveDatasetManager()

register_as_golden()

dm.register_as_golden(
    log_uuid="abc-123",
    note="Verified by QA",
    tags=["v2", "production"],
)

recommend_candidates()

candidates = dm.recommend_candidates(
    function_name="generate_response",
    limit=5,
)

VectorWaveAutoInjector

Auto-injection uses class methods — no instance creation needed.

configure()

from vectorwave import VectorWaveAutoInjector

# Set default configuration for all inject calls
VectorWaveAutoInjector.configure(
    auto=True,
    capture_return_value=True,
    team="ai-team",
)

inject()

# Inject @vectorize into all functions in a module
VectorWaveAutoInjector.inject(
    target_module_path="app.services.ai",
    recursive=False,          # Recursively scan submodules
    team="ai-team",           # Override config per inject call
)

Webhook Alerting

Alerts are configured via environment variables, not a function call:

# .env
ALERTER_STRATEGY=webhook                              # Enable webhook alerts
ALERTER_WEBHOOK_URL=https://discord.com/api/webhooks/...  # Webhook endpoint
ALERTER_MIN_LEVEL=ERROR                               # Minimum alert level
DRIFT_DETECTION_ENABLED=True                          # Enable drift alerts
DRIFT_DISTANCE_THRESHOLD=0.25                         # Drift threshold

Alert payloads are sent automatically when errors or drift events occur.