pip install abvdev

from abvdev import ABV
import re

def pii_masker(data: any, **kwargs) -> any:
    # Example: Simple email masking. Implement your more robust logic here.
    if isinstance(data, str):
        return re.sub(r"[a-zA-Z0-9_.+-]+@[a-zA-Z0-9-]+\.[a-zA-Z0-9-.]+", "[EMAIL_REDACTED]", data)
    elif isinstance(data, dict):
        return {k: pii_masker(data=v) for k, v in data.items()}
    elif isinstance(data, list):
        return [pii_masker(data=item) for item in data]
    return data

abv = ABV(
    mask=pii_masker,
    api_key="sk-abv-...", # your api key here
    host="https://app.abv.dev", # host="https://eu.app.abv.dev", for EU region
)

# Now, any input/output/metadata will be passed through pii_masker
with abv.start_as_current_span(name="user-query", input={"email": "[email protected]", "query": "..."}) as span:
    # The 'email' field in the input will be masked.
    pass

import logging

abv_logger = logging.getLogger("abv")
abv_logger.setLevel(logging.DEBUG)

from abvdev import ABV

# Sample approximately 20% of traces
abv_sampled = ABV(
    sample_rate=0.2,
    api_key="sk-abv-...", # your api key here
    host="https://app.abv.dev", # host="https://eu.app.abv.dev", for EU region
)

from abvdev import ABV

# Filter out database spans
abv = ABV(
    blocked_instrumentation_scopes=["sqlalchemy", "psycopg"],
    api_key="sk-abv-...", # your api key here
    host="https://app.abv.dev", # host="https://eu.app.abv.dev", for EU region
)

from opentelemetry.sdk.trace import TracerProvider

from abvdev import ABV

abv_tracer_provider = TracerProvider() # do not set to global tracer provider to keep isolation

abv = ABV(
    tracer_provider=abv_tracer_provider,
    api_key="sk-abv-...", # your api key here
    host="https://app.abv.dev", # host="https://eu.app.abv.dev", for EU region
)

abv.start_span(name="myspan").end() # Span will be isolated from remaining OTEL instrumentation

from concurrent.futures import ThreadPoolExecutor, as_completed
from abvdev import ABV, observe

abv = ABV(
    api_key="sk-abv-...", # your api key here
    host="https://app.abv.dev", # host="https://eu.app.abv.dev", for EU region
)
 
@observe
def execute_task(*args):
    return args
 
@observe
def execute_groups(task_args):
    trace_id = abv.get_current_trace_id()
    observation_id = abv.get_current_observation_id()
 
    with ThreadPoolExecutor(3) as executor:
        futures = [
            executor.submit(
                execute_task,
                *task_arg,
                abv_trace_id=trace_id,
                abv_parent_observation_id=observation_id,
            )
            for task_arg in task_args
        ]
 
        for future in as_completed(futures):
            future.result()
 
    return [f.result() for f in futures]
 
@observe()
def main():
    task_args = [["a", "b"], ["c", "d"]]
 
    execute_groups(task_args)
 
main()
 
abv.flush()

from abvdev import ABV

# Initialize clients for different projects
project_a_client = ABV(
    api_key="sk-abv-project-a-...",
    host="https://app.abv.dev"
    # or https://eu.app.abv.dev
)

project_b_client = ABV(
    api_key="sk-abv-project-b-...",
    host="https://app.abv.dev"
    # or https://eu.app.abv.dev
)

from abvdev import observe, get_client
 
@observe
def nested():
    # get_client call is context aware
    # if it runs inside another decorated function that has
    # abv_api_key passed, it does not need passing here again
    get_client().update_current_trace(user_id='myuser')
 
 
@observe
def process_data_for_project_a(data):
    # passing `abv_api_key` here again is not necessarily
    # as it is stored in execution context
    nested()
 
    return {"processed": data}
 
@observe
def process_data_for_project_b(data):
    # passing `abv_api_key` here again is not necessarily
    # as it is stored in execution context
    nested()
 
    return {"enhanced": data}
 
# Route to Project A
# Top-most decorated function needs `abv_api_key` kwarg
result_a = process_data_for_project_a(
    data="input data",
    abv_api_key="sk-abv-project-a-..."
)
 
# Route to Project B
# Top-most decorated function needs `abv_api_key` kwarg
result_b = process_data_for_project_b(
    data="input data",
    abv_api_key="sk-abv-project-b-..."
)

from abvdev import ABV
import datetime
import time
 
abv = ABV(
    api_key="sk-abv-...", # your api key here
    host="https://app.abv.dev", # host="https://eu.app.abv.dev", for EU region
)
 
# Start observation with specific type
with abv.start_as_current_observation(
    as_type="generation",
    name="TTFT-Generation"
) as generation:
 
    # simulate LLM time to first token
    time.sleep(3)
 
    # Update the generation with the time the model started to generate
    generation.update(
        completion_start_time=datetime.datetime.now(),
        output="some response",
    )
 
# Flush events in short-lived applications
abv.flush()

from abvdev import observe
 
# Tool calls to external services
@observe(as_type="tool")
def retrieve_context(query):
    results = vector_store.get(query)
    return results

from abvdev import get_client
abv = get_client()
 
def process_with_context_managers():
    with abv.start_as_current_observation(
        as_type="chain",
        name="retrieval-pipeline",
    ) as chain:
 
        # Retrieval step
        with abv.start_as_current_observation(
            as_type="retriever",
            name="vector-search",
        ) as retriever:
            search_results = perform_vector_search("user question")
            retriever.update(output={"results": search_results})