lumenova-beacon 2.6.3

Lumenova Beacon SDK - A Python SDK for observability tracing with OpenTelemetry-compatible span export

Lumenova Beacon SDK

A Python observability SDK for AI/LLM applications — trace agentic frameworks (LangChain, LangGraph, CrewAI, Strands), LLM calls, and custom code with OpenTelemetry-compatible spans.

Features

• LangChain/LangGraph Integration - Automatic tracing for chains, agents, tools, retrievers, with interrupt/resume and agent handoff support
• Strands Agents Integration - Callback handler for AWS Strands agent tracing
• CrewAI Integration - Event listener for CrewAI crew tracing
• LiteLLM Integration - Callback logger for LiteLLM proxy tracing
• Agentic Governance - Real-time policy enforcement for AI agent tool calls and LLM invocations
• System Probes - Run autonomous AI agents that probe your HTTP system locally (private APIs, custom auth) and produce scored markdown reports
• OpenTelemetry Integration - Automatic instrumentation for Anthropic, OpenAI, FastAPI, Redis, HTTPX, and more
• Manual & Decorator Tracing - Create spans manually or use @trace decorator
• Dataset Management - ActiveRecord-style API for managing test datasets
• Prompt Management - Version-controlled prompt templates with labels (staging, production)
• Experiment & Evaluation Management - Run experiments over datasets and evaluate results
• Data Masking - Built-in PII detection and redaction via Beacon Guardrails
• Flexible Transport - HTTP or file-based span export
• Full Async Support - Async/await throughout

Requirements

• Python 3.10+

Installation

# Base installation
pip install lumenova-beacon

# With OpenTelemetry support
pip install lumenova-beacon[opentelemetry]

# With LangChain/LangGraph support
pip install lumenova-beacon[langchain]

# With LiteLLM support
pip install lumenova-beacon[litellm]

# With Strands Agents support
pip install lumenova-beacon[strands]

# With CrewAI support
pip install lumenova-beacon[crewai]

# With AWS Secrets Manager support
pip install lumenova-beacon[aws]

Quick Start

LangChain / LangGraph

from lumenova_beacon import BeaconClient, BeaconLangGraphHandler
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate

# Initialize client
client = BeaconClient(
    endpoint="https://your-beacon-endpoint.lumenova.ai",
    api_key="your-api-key",
)

# Create a tracing handler
handler = BeaconLangGraphHandler(session_id="session-123")

# All LangChain operations are now traced automatically
llm = ChatOpenAI(model="gpt-4")
prompt = ChatPromptTemplate.from_template("Tell me about {topic}")
chain = prompt | llm

response = chain.invoke(
    {"topic": "AI agents"},
    config={"callbacks": [handler]}
)

Basic Tracing

from lumenova_beacon import BeaconClient, trace

client = BeaconClient(
    endpoint="https://your-beacon-endpoint.lumenova.ai",
    api_key="your-api-key",
    session_id="my-session"
)

@trace
def my_function(x, y):
    return x + y

result = my_function(10, 20)  # Automatically traced

No endpoint? Pass file_directory='./traces' instead of endpoint to write spans as JSON locally — useful for development and tests. See File Transport.

Configuration

Environment Variables

All environment variables work as fallback — constructor parameters override them:

Variable	Purpose	Default
BEACON_ENDPOINT	API base URL for OTLP export	(required unless using file_directory)
BEACON_API_KEY	Authentication token
BEACON_SESSION_ID	Default session ID for spans
BEACON_SERVICE_NAME	Service name for OTEL resource (fallback: OTEL_SERVICE_NAME)
BEACON_ENVIRONMENT	Deployment environment (e.g., "production", "staging")
BEACON_USER_EMAIL	Default user email for attribution
BEACON_USER_NAME	Default user display name for attribution
BEACON_USER_ID	Default user ID for attribution
BEACON_VERIFY	SSL certificate verification	true
BEACON_EAGER_EXPORT	Export spans eagerly on end	true
BEACON_ISOLATED	Use a private TracerProvider — set to true if your app already runs another OTEL exporter (e.g. infra observability) so Beacon spans don't collide with it	false
BEACON_EXTRA_OTLP_ENDPOINTS	Additional OTLP gRPC endpoints (comma-separated)
BEACON_AWS_SECRET_NAME	AWS Secrets Manager secret name (resolves API key)
BEACON_AWS_SECRET_KEY	Key path within AWS secret	api_key
BEACON_AWS_REGION	AWS region for Secrets Manager	(boto3 default)

# Bash/Linux/macOS
export BEACON_ENDPOINT="https://your-beacon-endpoint.lumenova.ai"
export BEACON_API_KEY="your-api-key"
export BEACON_SESSION_ID="my-session"

# PowerShell
$env:BEACON_ENDPOINT = "https://your-beacon-endpoint.lumenova.ai"
$env:BEACON_API_KEY = "your-api-key"
$env:BEACON_SESSION_ID = "my-session"

Configuration Options

from lumenova_beacon import BeaconClient

client = BeaconClient(
    # Connection
    endpoint="https://your-beacon-endpoint.lumenova.ai",
    api_key="your-api-key",
    verify=True,

    # Span Configuration
    session_id="my-session",
    service_name="my-service",
    environment="production",

    # User Identity
    user_email="[e-mail]",
    user_name="Alice",
    user_id="user-123",

    # OpenTelemetry
    auto_instrument_opentelemetry=True,   # Auto-configure OTEL (default: True)
    isolated=False,                        # Use private TracerProvider (default: False)
    auto_instrument_litellm=False,         # Auto-configure LiteLLM (default: False)

    # Data Masking
    masking_function=None,                 # Custom masking function (optional)

    # General
    eager_export=True,
    record_stacktrace=False,               # Capture stack traces on exceptions (default: False)

    # Additional options passed via **kwargs
    # enabled=True,                        # Enable/disable tracing
    # headers={"Custom-Header": "value"},  # Custom HTTP headers
)

File Transport

For local development or testing, use file_directory instead of endpoint:

from lumenova_beacon import BeaconClient

client = BeaconClient(
    file_directory="./traces",
)

Integrations

1. LangChain/LangGraph

Automatically trace all LangChain and LangGraph operations — chains, agents, tools, retrievers, and LLM calls:

from lumenova_beacon import BeaconClient, BeaconLangGraphHandler
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate

client = BeaconClient()
handler = BeaconLangGraphHandler(session_id="session-123")

# Use with request-time callbacks (recommended)
llm = ChatOpenAI(model="gpt-4")
response = llm.invoke(
    "What is the capital of France?",
    config={"callbacks": [handler]}
)

# Works with chains
prompt = ChatPromptTemplate.from_template("Tell me about {topic}")
chain = prompt | llm
response = chain.invoke(
    {"topic": "AI"},
    config={"callbacks": [handler]}
)

# Traces agents, tools, retrievers, and more
from langchain.agents import create_react_agent, AgentExecutor

agent = create_react_agent(llm, tools, prompt)
executor = AgentExecutor(agent=agent, tools=tools)
result = executor.invoke(
    {"input": "What's the weather?"},
    config={"callbacks": [handler]}
)

LangGraph with Interrupt/Resume

For LangGraph agents with checkpointing, use BeaconLangGraphConfig to automatically continue traces across interrupt/resume cycles:

from lumenova_beacon import BeaconClient, BeaconLangGraphConfig

client = BeaconClient()

beacon = BeaconLangGraphConfig(
    graph=agent.graph,
    thread_id=thread_id,
    agent_name="planner",
    session_id="session-123",
    autodetect_handoffs=True,  # Auto-detect agent-to-agent handoffs
)
config = beacon.get_config()
result = await agent.graph.ainvoke(state, config)

# For async checkpointers (e.g., AsyncPostgresSaver):
config = await beacon.aget_config()

2. LiteLLM

Trace all LiteLLM operations across multiple LLM providers:

from lumenova_beacon import BeaconClient, BeaconLiteLLMLogger
import litellm

# Option 1: Auto-instrumentation
client = BeaconClient(auto_instrument_litellm=True)

# Option 2: Manual registration
client = BeaconClient()
litellm.callbacks = [BeaconLiteLLMLogger()]

# All LiteLLM calls are now traced
response = litellm.completion(
    model="gpt-4",
    messages=[{"role": "user", "content": "Hello"}],
    metadata={
        "generation_name": "greeting",
        "session_id": "user-123",
    }
)

3. Strands Agents

Trace AWS Strands Agent executions with automatic span hierarchy:

from lumenova_beacon import BeaconClient, BeaconStrandsHandler
from strands import Agent

client = BeaconClient()
handler = BeaconStrandsHandler(
    session_id="my-session",
    agent_name="My Agent",
)

agent = Agent(model=model, callback_handler=handler)
result = agent("Hello, world!")
print(handler.trace_id)  # Link to Beacon trace

4. CrewAI

Trace CrewAI Crew executions via the event listener:

from lumenova_beacon import BeaconClient, BeaconCrewAIListener
from crewai import Agent, Crew, Task

client = BeaconClient()

# Auto-registers with CrewAI event bus
listener = BeaconCrewAIListener(
    session_id="my-session",
    crew_name="My Research Crew",
)

crew = Crew(agents=[...], tasks=[...])
result = crew.kickoff()
print(listener.trace_id)  # Link to Beacon trace

5. OpenTelemetry

Beacon automatically configures OpenTelemetry to export spans:

from lumenova_beacon import BeaconClient
from opentelemetry.instrumentation.anthropic import AnthropicInstrumentor
from opentelemetry.instrumentation.openai import OpenAIInstrumentor

# Initialize (auto-configures OpenTelemetry)
client = BeaconClient(
    endpoint="https://your-beacon-endpoint.lumenova.ai",
    api_key="your-api-key",
    auto_instrument_opentelemetry=True  # Default
)

# Instrument libraries
AnthropicInstrumentor().instrument()
OpenAIInstrumentor().instrument()

# Now all API calls are automatically traced!
from anthropic import Anthropic
anthropic = Anthropic()
response = anthropic.messages.create(
    model="claude-3-5-sonnet-20241022",
    messages=[{"role": "user", "content": "Hello!"}]
)  # Automatically traced with proper span hierarchy

Supported Instrumentors

Install additional instrumentors as needed:

pip install opentelemetry-instrumentation-anthropic
pip install opentelemetry-instrumentation-openai
pip install opentelemetry-instrumentation-fastapi
pip install opentelemetry-instrumentation-redis
pip install opentelemetry-instrumentation-httpx
pip install opentelemetry-instrumentation-requests

Tracing

Decorator Tracing

The @trace decorator automatically captures function execution:

from lumenova_beacon import trace

# Simple usage
@trace
def process_data(data):
    return data.upper()

# With custom name
@trace(name="custom_operation")
def another_function():
    pass

# Capture inputs and outputs
@trace(capture_args=True, capture_result=True)
def calculate(x, y):
    return x + y

# Works with async functions
@trace
async def async_operation():
    await some_async_call()

Manual Tracing

For more control, use context managers:

from lumenova_beacon import BeaconClient
from lumenova_beacon.types import SpanKind, StatusCode

client = BeaconClient()

# Context manager
with client.trace("operation_name") as span:
    span.set_attribute("user_id", "123")
    span.set_input({"query": "search term"})

    try:
        result = do_work()
        span.set_output(result)
        span.set_status(StatusCode.OK)
    except Exception as e:
        span.record_exception(e)
        span.set_status(StatusCode.ERROR, str(e))
        raise

# Async context manager
async with client.trace("async_operation") as span:
    result = await async_work()
    span.set_output(result)

# Direct span creation
span = client.create_span(
    name="manual_span",
    kind=SpanKind.CLIENT,
)
span.start()
# ... do work ...
span.end()

Sessions

A session_id groups related traces — a multi-turn conversation, a batch run, a single user journey — so they can be filtered together in Beacon. Set it once on the client (or via BEACON_SESSION_ID) and it's inherited by every span created through the client. You can also override it per integration handler (BeaconLangGraphHandler(session_id=...), BeaconStrandsHandler(session_id=...), etc.) or per span via span.set_attribute("session_id", "...").

Agentic Governance

Enforce governance policies on AI agent actions in real time. The governance system evaluates tool calls and LLM invocations against policy stacks before and after execution, blocking actions that violate your rules.

Key capabilities:

• Pre & post execution hooks — evaluate actions before they run (prevent violations) and after (audit outputs)
• Fail-open resilience — defaults to allowing actions when the governance API is unreachable
• Violation strategies — raise immediately, stop gracefully after current action, or escalate after N violations
• Content reinjection — policies that emit a top-level output string (e.g. PII-masked tool output) can have that replacement applied automatically via @governance, handler.wrap(...) on a BeaconLangGraphGovernanceHandler instance (see Wrapping for content reinjection below), or the LangGraph helpers wrap_tool_node / wrap_chat_model. Callback-only attachment observes but does not substitute.

Decorator

from lumenova_beacon import governance

# Bare decorator — requires GovernanceConfig on the active BeaconClient
@governance
def search_web(query: str) -> str:
    return web_search(query)

# With policy stack IDs
@governance(stack_ids=["my-policy-stack"])
def send_email(to: str, body: str) -> bool:
    return email_client.send(to, body)

# Tag-based policy discovery with strict mode
@governance(tags=["env:prod", "team:security"], fail_open=False)
def delete_record(record_id: str) -> None:
    db.delete(record_id)

# Works with async functions
@governance(stack_ids=["my-policy-stack"])
async def run_query(sql: str) -> list[dict]:
    return await db.execute(sql)

LangChain/LangGraph Callback Handler

from lumenova_beacon import BeaconLangGraphHandler, BeaconLangGraphGovernanceHandler
from langchain_openai import ChatOpenAI

handler = BeaconLangGraphHandler(session_id="session-123")
governance_handler = BeaconLangGraphGovernanceHandler(
    stack_ids=["my-policy-stack"],
    fail_open=True,
    on_violation="raise",
)

llm = ChatOpenAI(model="gpt-4")
response = llm.invoke(
    "What is the capital of France?",
    config={"callbacks": [handler, governance_handler]}
)

Wrapping for content reinjection

Use handler.wrap(...) (instead of callback-only attachment) when a policy returns a top-level output field that should actually replace the tool arguments, tool return value, LLM prompt, or LLM response. Callbacks can only observe and block; wrap rebinds the inner callable so the substitution takes effect.

wrap is an instance method — construct a handler first, then call .wrap(...) on the instance.

from langchain_core.tools import tool
from langchain_openai import ChatOpenAI
from lumenova_beacon import BeaconLangGraphGovernanceHandler

@tool
def read_file(path: str) -> str:
    """Read a file."""
    with open(path) as f:
        return f.read()

# 1. Construct the handler — it owns shared violation state.
handler = BeaconLangGraphGovernanceHandler(
    stack_ids=["pii-redaction-stack"],
    fail_open=True,
    on_violation="raise",
)

# 2. Wrap on the INSTANCE. Mutates in place; returns the same target.
wrapped_tool = handler.wrap(read_file)                          # BaseTool
wrapped_tools = handler.wrap([read_file])                       # list[BaseTool]
wrapped_llm = handler.wrap(ChatOpenAI(model="gpt-4o-mini"))     # BaseChatModel

# 3. All wrapped targets share handler.violation_count, the stopped flag,
#    and the invocation chain via closure.
print(handler.violation_count)

Notes:

• wrap accepts a BaseTool, a Sequence[BaseTool], or a BaseChatModel.
• The input is mutated in place: BaseTool.func/coroutine and BaseChatModel.invoke/ainvoke are rebound. Keep a reference before wrapping if you need an unwrapped copy.
• Idempotent: wrapping an already-wrapped target is a no-op.
• stream / astream on a wrapped chat model bypass governance — call invoke / ainvoke instead.
• Tools that override _run/_arun directly (no func/coroutine attribute) cannot be wrapped — apply @governance to the inner function before building the tool, or expose func/coroutine.

A full LangGraph StateGraph example using this pattern lives in the SDK source tree at examples/llm_integrations/langchain/langchain_governance_wrap.py.

Integrated Tracing + Governance with BeaconLangGraphConfig

For LangGraph agents, pass a GovernanceConfig to BeaconLangGraphConfig to get both tracing and governance in a single setup:

from lumenova_beacon import BeaconLangGraphConfig, GovernanceConfig

beacon = BeaconLangGraphConfig(
    graph=agent.graph,
    thread_id=thread_id,
    agent_name="planner",
    governance=GovernanceConfig(
        stack_ids=["my-policy-stack"],
        fail_open=True,
        on_violation="raise",
        max_violations=3,
    ),
)
config = beacon.get_config()
result = await agent.graph.ainvoke(state, config)

GovernanceConfig Options

from lumenova_beacon import GovernanceConfig

config = GovernanceConfig(
    stack_ids=["stack-1", "stack-2"],       # Policy stack IDs (at least one of stack_ids/tags required)
    tags=["env:prod"],                       # Tag-based policy discovery
    fail_open=True,                          # Allow actions when API is unreachable (default: True)
    enabled_hooks={"pre_tool", "post_tool"}, # Which hooks to run (default: all four)
    on_violation="raise",                    # "raise" (immediate) or "stop" (graceful shutdown)
    max_violations=5,                        # Escalate to stop mode after N violations
    timeout=2.0,                             # API call timeout in seconds
    debug=False,                             # Enable debug logging
)

Available hooks: pre_tool, post_tool, pre_llm, post_llm.

Handling Violations

from lumenova_beacon.exceptions import GovernanceViolationError

try:
    result = agent.invoke(state, config)
except GovernanceViolationError as e:
    print(e.message)       # Human-readable reason
    print(e.policies)      # List of rules that were evaluated
    print(e.latency_ms)    # Evaluation latency

System Probes

Run autonomous AI agents that probe your HTTP system, classify each outcome, and produce a scored markdown report. The agent (LangGraph + LLM + scoring) runs on Beacon; the SDK only claims the run, polls for HTTP requests the agent wants executed, runs each one against your target with your local credentials, and PATCHes the response back. In SDK mode, Beacon never sees your target's URL or credentials.

Use a probe when you want broad coverage of an HTTP API without writing tests by hand, when you want adversarial probing (e.g. OWASP LLM Top 10) against an LLM-backed endpoint, or when your API lives behind a VPN / private VPC where Beacon can't reach it directly.

Configure the run from the Beacon UI (target system → probe evaluator → click Run probe), then run it locally:

Path A — built-in HTTP dispatcher

Pass auth as a kwarg matching the target's auth_scheme_hint. The SDK fires httpx itself and injects the auth header on every outbound request.

from lumenova_beacon import Probe

# Bearer auth (auth_scheme_hint='bearer')
Probe.run("<run-id-from-the-ui>", bearer_token="<your-bearer-token>")

# Custom header auth (e.g. auth_scheme_hint='custom_headers')
Probe.run("<run-id-from-the-ui>", headers={"X-API-Key": "<your-api-key>"})

# No auth (auth_scheme_hint='none')
Probe.run("<run-id-from-the-ui>")

Probe.run blocks until the run reaches a terminal status. Use Probe.arun(...) for the async variant.

Path B — custom callable

Own the connection yourself — for custom auth flows, non-HTTP transports, or mocks.

import time
from lumenova_beacon import Probe, ProbeHttpRequest, ProbeHttpSuccessResponse

def my_dispatcher(req: ProbeHttpRequest) -> ProbeHttpSuccessResponse:
    started = time.monotonic()
    response = my_internal_client.send(
        req.method,
        req.url,
        headers=req.request_headers,
        body=req.request_body,
    )
    return ProbeHttpSuccessResponse(
        duration_ms=int((time.monotonic() - started) * 1000),
        status_code=response.status_code,
        response_headers=dict(response.headers),
        response_body=response.text,
    )

Probe.run("<run-id-from-the-ui>", target_callable=my_dispatcher)

Mixing target_callable with an auth kwarg raises immediately at Probe.run entry — pick one.

The SDK uses the same BEACON_ENDPOINT + BEACON_API_KEY env vars (or BeaconClient configuration) as the rest of the SDK to reach Beacon.

Dataset Management

Manage test datasets with an ActiveRecord-style API. All methods have async variants with an a prefix (e.g., acreate, aget, alist).

from lumenova_beacon import BeaconClient
from lumenova_beacon.datasets import Dataset, DatasetRecord

client = BeaconClient()

# Create dataset
dataset = Dataset.create(
    name="qa-evaluation",
    description="Question answering test cases"
)

# Add records with flexible column-based data
dataset.create_record(
    data={
        "prompt": "What is AI?",
        "expected_answer": "Artificial Intelligence is...",
        "difficulty": "easy",
        "category": "definitions"
    }
)

# Bulk create records
dataset.bulk_create_records([
    {"data": {"question": "What is ML?", "expected_answer": "Machine Learning..."}},
    {"data": {"question": "What is DL?", "expected_answer": "Deep Learning..."}},
])

# List, get, update, delete
datasets, pagination = Dataset.list(page=1, page_size=20, search="qa")
dataset = Dataset.get(dataset_id="dataset-uuid", include_records=True)
dataset.update(name="updated-name", description="New description")
dataset.delete()

Prompt Management

Version-controlled prompt templates with labels. All methods have async variants with an a prefix.

from lumenova_beacon import BeaconClient
from lumenova_beacon.prompts import Prompt

client = BeaconClient()

# Create text prompt
prompt = Prompt.create(
    name="greeting",
    template="Hello {{name}}! Welcome to {{company}}.",
    description="Customer greeting template",
    tags=["customer-support", "greeting"]
)

# Create chat prompt
prompt = Prompt.create(
    name="support-bot",
    messages=[
        {"role": "system", "content": "You are a helpful assistant for {{product}}."},
        {"role": "user", "content": "{{question}}"}
    ],
)

# Get latest version (sync)
prompt = Prompt.get(name="greeting")

# Get specific version (async)
prompt = await Prompt.aget(name="greeting", version=2)

# Get labeled version (sync)
prompt = Prompt.get(name="greeting", label="production")

# Get by ID (async)
prompt = await Prompt.aget(prompt_id="prompt-uuid")

# Format prompt with variables
message = prompt.format(name="Alice", company="Acme Corp")
# Result: "Hello Alice! Welcome to Acme Corp."

# Versioning and labels
new_version = prompt.publish(
    template="Hi {{name}}! Welcome to {{company}}. We're excited to have you!",
    message="Added enthusiastic tone"
)
prompt.set_label("production", version=2)

# Convert to LangChain template
lc_prompt = prompt.to_langchain()  # Returns PromptTemplate or ChatPromptTemplate

LangChain Conversion

from langchain_core.prompts import PromptTemplate, ChatPromptTemplate

# Convert text prompt to LangChain (sync)
prompt = Prompt.get(name="greeting", label="production")
lc_prompt = prompt.to_langchain()  # Returns PromptTemplate
result = lc_prompt.format(name="Bob", company="TechCorp")

# Convert chat prompt to LangChain (async)
chat_prompt = await Prompt.aget("support-bot", label="production")
lc_chat = chat_prompt.to_langchain()  # Returns ChatPromptTemplate
messages = lc_chat.format_messages(product="DataHub", question="Reset password?")

# Use in chain
from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-4")
chain = lc_chat | llm
response = await chain.ainvoke({"product": "CloudSync", "question": "Why sync failing?"})

List and Search

# List all prompts (sync)
prompts = Prompt.list(page=1, page_size=20)

# Filter by tags (async)
support_prompts = await Prompt.alist(tags=["customer-support"])

# Search by text (sync)
results = Prompt.list(search="greeting")

# Async version
prompts = await Prompt.alist(page=1, page_size=10)

Experiment Management

Experiments run a pipeline of steps (PROMPT, LLM, EVALUATION, EXTERNAL_AGENT) over a dataset. The server orchestrates the run and stores results.

External Agents

Use an EXTERNAL_AGENT step to plug your own code — a LangGraph graph, a RAG pipeline, anything callable — into a pipeline step. The server runs every other step it can, then transitions the experiment to WAITING_FOR_EXTERNAL; the SDK polls for pending inputs, dispatches each one to your callable, and writes the result back.

from lumenova_beacon.experiments import (
    Experiment, ExperimentStep, ExperimentStepType, ExperimentStatus,
)

experiment = Experiment.create(
    name="agent-benchmark",
    dataset_id="dataset-uuid",
    steps=[
        ExperimentStep(
            step_type=ExperimentStepType.EXTERNAL_AGENT,
            output_column="agent_output",
            config={
                "agent_name": "My Agent",
                "variable_mappings": {"input": "question"},
            },
        ),
    ],
)

# Blocks until every pending external run has been dispatched.
# `my_agent_fn` receives the resolved input dict and returns the value
# to write into `output_column`.
experiment.start(external_agents={"My Agent": my_agent_fn})

# Resume a paused experiment from the same callable.
if experiment.status == ExperimentStatus.WAITING_FOR_EXTERNAL:
    experiment.continue_experiment(external_agents={"My Agent": my_agent_fn})

Evaluation Management

Evaluations score outputs using one of three evaluator types — LLM-as-judge, code-based, or agent-as-judge — and come in two flavors selected by which source you pass:

• Trace-based (filter_rules=...): scores spans matching a filter. Set active=True and the evaluator runs continuously on new traces.
• Dataset-based (dataset_id=...): scores rows in a dataset on demand.

from lumenova_beacon.evaluations import Evaluation, Evaluator

# Reuse an existing evaluator configured in the Beacon UI
evaluator = Evaluator.get(evaluator_id="evaluator-uuid")

# Trace-based: auto-runs on new spans matching the filter
evaluation = Evaluation.create(
    name="prod-quality-monitor",
    evaluator_id=evaluator.id,
    variable_mappings={
        "question": {"data_mode": "reduced", "reduced_fields": ["input"]},
        "answer":   {"data_mode": "reduced", "reduced_fields": ["output"]},
    },
    filter_rules={"logic": "AND", "rules": [
        {"field": "span.name", "operator": "contains", "value": "chat"},
    ]},
    active=True,
)

# Dataset-based: scores rows in a dataset (run on demand)
evaluation = Evaluation.create(
    name="qa-accuracy",
    evaluator_id=evaluator.id,
    variable_mappings={
        "question": {"data_mode": "reduced", "column": "input"},
        "answer":   {"data_mode": "reduced", "column": "output"},
    },
    dataset_id="dataset-uuid",
)

Extraction Engines for Variable Mappings

Each variable mapping picks an extraction engine for the field expression. Three are available:

• JSONPath — straightforward path access ($.input.messages[0].content). Fast and familiar; right for a single key or index.
• JSONata — structural transforms (filter by predicate, join, aggregate, reshape).
• Python — anything that needs general-purpose code (regex, multi-step helpers, NumPy). The expression must define a top-level def extract(data): function.

Engines can be chained as a pipeline (e.g. JSONata → Python) when a single expression isn't enough. Caps mirrored client-side: each expr is at most 64 KiB and a pipeline is at most 32 steps.

from lumenova_beacon import VariableMapping, ProcessorStep, ExtractionEngine, ProcessorKind

# JSONPath (legacy `json_path` field — also accepted as a raw dict shape).
question = VariableMapping.jsonpath("$.messages[0].content")

# JSONata for filter/join/aggregate/reshape transforms.
answer = VariableMapping.jsonata("$join(messages[role='assistant'].content, ' ')")

# Python for general-purpose transforms.
summary = VariableMapping.python(
    "def extract(data):\n"
    "    return data['output'].strip().lower()\n"
)

# Multi-step pipeline (JSONata → Python).
chained = VariableMapping(
    data_mode="raw",
    pipeline=[
        ProcessorStep(engine=ExtractionEngine.JSONATA, expr="messages.content"),
        ProcessorStep(
            engine=ExtractionEngine.PYTHON,
            kind=ProcessorKind.RESHAPE,
            expr="def extract(data):\n    return ' '.join(data)\n",
        ),
    ],
)

evaluation = Evaluation.create(
    name="multi-engine",
    evaluator_id=evaluator.id,
    variable_mappings={"question": question, "answer": answer, "summary": summary},
    filter_rules={"logic": "AND", "rules": []},
)

The legacy dict[str, Any] shape ({"question": {"data_mode": "raw", "json_path": "$.x"}}) keeps working unchanged. Validation runs at construction time: incompatible engine/kind pairs (e.g. jsonpath + filter), oversize expressions, and the pipeline-with-json_path anti-pattern raise ValueError before the request hits the wire. Requires the Beacon backend version that ships the Python extraction engine (lumenova-guardrails MR !337) for the python and pipeline shapes; older backends accept JSONPath only.

LLM Config Management

from lumenova_beacon.llm_configs import LLMConfig

config = LLMConfig.get(config_id="config-uuid")
configs, pagination = LLMConfig.list(page=1, page_size=20)

Data Masking

Automatically mask sensitive data (PII) before spans are exported:

from lumenova_beacon import BeaconClient
from lumenova_beacon.masking.integrations.beacon_guardrails import (
    create_beacon_masking_function,
    MaskingMode,
    PIIType,
)

# Create a masking function backed by Beacon Guardrails API
masking_fn = create_beacon_masking_function(
    pii_types=[PIIType.PERSON, PIIType.EMAIL_ADDRESS, PIIType.US_SSN],
    mode=MaskingMode.REDACT,
)

# Pass it to the client - all span data is masked before export
client = BeaconClient(
    endpoint="https://your-beacon-endpoint.lumenova.ai",
    api_key="your-api-key",
    masking_function=masking_fn,
)

You can also provide a custom masking function:

def my_masking_fn(text: str) -> str:
    return text.replace("secret", "***")

client = BeaconClient(masking_function=my_masking_fn)

Guardrails

Apply content guardrail policies via the Beacon API:

from lumenova_beacon.guardrails import Guardrail

guardrail = Guardrail(guardrail_id="guardrail-uuid")

# Sync
result = guardrail.apply("some user input")

# Async
result = await guardrail.aapply("some user input")

Optional metadata

Pass an optional metadata dict to enable retrieval/grounding-aware policies:

result = guardrail.apply(
    "Paris is the capital of France.",
    metadata={
        "query": "What is the capital of France?",
        "context": [
            "Paris is the capital and largest city of France.",
            "France is a country in Western Europe.",
        ],
    },
)

• metadata is optional — omit it entirely if you only need to validate text. Both query and context are also individually optional within metadata.
• query is the user's original question; it is forwarded to evaluation policies that score relevance/grounding against the model output.
• context is the list of retrieved document texts the model used to answer (typical RAG flow). Pass the actual content of each chunk, not links to it. Each entry is sanitized alongside the main text and round-trips into the response (outputs[].input.metadata.context, outputs[].output.metadata.context).

API Reference

Main Exports

from lumenova_beacon import (
    BeaconClient,           # Main client
    BeaconConfig,           # Configuration class
    get_client,             # Get current client singleton
    trace,                  # Tracing decorator
    # Integrations (lazy-loaded)
    BeaconLangGraphHandler,  # LangChain/LangGraph
    BeaconLangGraphConfig,  # LangGraph configuration (adds support for interruptions)
    BeaconStrandsHandler,   # Strands Agents
    BeaconCrewAIListener,   # CrewAI
    BeaconLiteLLMLogger,    # LiteLLM
    # Governance (lazy-loaded)
    BeaconLangGraphGovernanceHandler,  # LangChain/LangGraph governance
    GovernanceConfig,                  # Governance configuration
)

from lumenova_beacon.governance import governance  # Function decorator

from lumenova_beacon.datasets import Dataset, DatasetRecord
from lumenova_beacon.prompts import Prompt
from lumenova_beacon.experiments import Experiment
from lumenova_beacon.evaluations import Evaluation, EvaluationRun
from lumenova_beacon.llm_configs import LLMConfig
from lumenova_beacon.guardrails import Guardrail
from lumenova_beacon.types import SpanKind, StatusCode, SpanType

BeaconClient

client = BeaconClient(
    endpoint: str | None = None,
    api_key: str | None = None,
    file_directory: str | None = None,
    session_id: str | None = None,
    user_email: str | None = None,
    user_name: str | None = None,
    user_id: str | None = None,
    service_name: str | None = None,
    environment: str | None = None,
    auto_instrument_opentelemetry: bool = True,
    isolated: bool = False,
    auto_instrument_litellm: bool = False,
    masking_function: Callable | None = None,
    verify: bool | None = None,
    eager_export: bool | None = None,
    record_stacktrace: bool = False,
)

# Methods
span = client.create_span(name, kind, span_type, session_id)
ctx = client.trace(name, kind, span_type)  # Context manager (sync & async)
client.export_span(span)     # Export a single span
client.export_spans(spans)   # Export multiple spans
client.flush()               # Flush pending spans

Dataset

# Class methods (sync - simple names)
dataset = Dataset.create(name: str, description: str | None = None, column_schema: list[dict[str, Any]] | None = None)
dataset = Dataset.get(dataset_id: str, include_records: bool = False)
datasets, pagination = Dataset.list(page=1, page_size=20, search=None)

# Class methods (async - 'a' prefix)
dataset = await Dataset.acreate(...)
dataset = await Dataset.aget(...)
datasets, pagination = await Dataset.alist(...)

# Instance methods (sync - simple names)
dataset.save()
dataset.update(name=None, description=None)
dataset.delete()
record = dataset.create_record(data: dict[str, Any])
dataset.bulk_create_records(records: list[dict])
records, pagination = dataset.list_records(page=1, page_size=50)

# Instance methods (async - 'a' prefix)
await dataset.asave()
await dataset.aupdate(...)
await dataset.adelete()
record = await dataset.acreate_record(...)
await dataset.abulk_create_records(...)
records, pagination = await dataset.alist_records(...)

# Properties
dataset.id
dataset.name
dataset.description
dataset.record_count
dataset.created_at
dataset.updated_at
dataset.column_schema

DatasetRecord

# Class methods (sync - simple names)
record = DatasetRecord.get(dataset_id: str, record_id: str)
records, pagination = DatasetRecord.list(dataset_id: str, page=1, page_size=50)

# Class methods (async - 'a' prefix)
record = await DatasetRecord.aget(...)
records, pagination = await DatasetRecord.alist(...)

# Instance methods (sync - simple names)
record.save()
record.update(data: dict[str, Any] | None = None)
record.delete()

# Instance methods (async - 'a' prefix)
await record.asave()
await record.aupdate(...)
await record.adelete()

# Properties
record.id
record.dataset_id
record.data  # dict[str, Any] - flexible column data
record.created_at
record.updated_at

Prompt

# Class methods (sync - simple names)
prompt = Prompt.create(name, template=None, messages=None, description=None, tags=None)
prompt = Prompt.get(prompt_id=None, *, name=None, label="latest", version=None)
prompts = Prompt.list(page=1, page_size=10, tags=None, search=None)

# Class methods (async - 'a' prefix)
prompt = await Prompt.acreate(...)
prompt = await Prompt.aget(...)
prompts = await Prompt.alist(...)

# Instance methods (sync - simple names)
prompt.update(name=None, description=None, tags=None)
prompt.delete()
new_version = prompt.publish(template=None, messages=None, message="")
prompt.set_label(label: str, version: int | None = None)

# Instance methods (async - 'a' prefix)
await prompt.aupdate(...)
await prompt.adelete()
new_version = await prompt.apublish(...)
await prompt.aset_label(...)

# Rendering (always sync)
result = prompt.format(**kwargs)
result = prompt.compile(variables: dict)
template = prompt.to_template()  # Convert to Python f-string format
lc_prompt = prompt.to_langchain()  # Convert to LangChain template

# Properties
prompt.id
prompt.name
prompt.type  # "text" or "chat"
prompt.version
prompt.template  # For TEXT prompts
prompt.messages  # For CHAT prompts
prompt.labels  # list[str]
prompt.tags  # list[str]

Span

span = Span(name, kind, span_type)

# Lifecycle
span.start()
span.end(status_code=StatusCode.OK)

# Status
span.set_status(StatusCode.ERROR, "description")
span.record_exception(exc: Exception)

# Attributes
span.set_attribute("key", value)
span.set_attributes({"k1": "v1", "k2": "v2"})
span.set_input(data: dict)
span.set_output(data: dict)
span.set_metadata("key", value)

# Properties
span.trace_id
span.span_id
span.parent_id
span.name
span.kind
span.span_type

Type Enums

from lumenova_beacon.types import SpanKind, StatusCode, SpanType

# SpanKind
SpanKind.INTERNAL
SpanKind.SERVER
SpanKind.CLIENT
SpanKind.PRODUCER
SpanKind.CONSUMER

# StatusCode
StatusCode.UNSET
StatusCode.OK
StatusCode.ERROR

# SpanType
SpanType.SPAN
SpanType.GENERATION
SpanType.CHAIN
SpanType.TOOL
SpanType.RETRIEVAL
SpanType.AGENT
SpanType.FUNCTION
SpanType.REQUEST
SpanType.SERVER
SpanType.TASK
SpanType.CACHE
SpanType.EMBEDDING
SpanType.HANDOFF
SpanType.CONDITIONAL

Error Handling

All exceptions inherit from BeaconError. Key exception types:

• ConfigurationError — invalid configuration
• TransportError (HTTPTransportError, FileTransportError) — export failures
• GovernanceViolationError — policy blocked an action (includes message, policies, latency_ms)
• DatasetError, PromptError, ExperimentError, EvaluationError — resource-specific errors (each with NotFound and Validation variants)

Retry Logic

All HTTP operations automatically retry up to 3 times with exponential backoff:

from lumenova_beacon.exceptions import PromptNetworkError

try:
    prompt = Prompt.get(name="my-prompt")
except PromptNetworkError as e:
    # Failed after 3 automatic retries
    print(f"Network error: {e}")
except PromptNotFoundError as e:
    # Prompt doesn't exist
    print(f"Not found: {e}")

Graceful Degradation

from lumenova_beacon import BeaconClient

# Disable tracing in development
client = BeaconClient(enabled=False)

# Tracing becomes no-op when disabled
@trace
def my_function():
    return "result"  # No tracing overhead

License

This project is licensed under the Apache License 2.0 - see the LICENSE file for details.