@guardrail_span

Wraps a guardrail or validation function with an OpenInference GUARDRAIL span. Captures the guardrail name, pass/fail result, and an optional confidence score.

Decorator only

@guardrail_span is a decorator only -- there is no context manager alternative. The pass/fail logic is determined automatically: if the function returns normally, the result is "pass"; if it raises an exception, the result is "fail".

---

Signature

PythonTypeScript

def guardrail_span(
    name: str | None = None,
    guardrail_name: str | None = None,
) -> Callable

interface GuardrailSpanOptions {
    name?: string;
    guardrailName?: string;
}

function guardrailSpan<A extends unknown[], R>(
    options: GuardrailSpanOptions,
    fn: (...args: A) => R | Promise<R>,
): (...args: A) => Promise<R>

---

Parameters

Parameter	Type	Default	Description
name	str \| None	None	Span name. Defaults to the decorated function's name.
guardrail_name	str \| None	None	Logical guardrail name recorded as an attribute. Defaults to the span name (which defaults to the function name).

---

Span Attributes

Attribute	Value
openinference.span.kind	"GUARDRAIL"
guardrail.name	Logical guardrail name (from guardrail_name, or name, or function name)
guardrail.result	"pass" if the function returns without raising; "fail" if it raises an exception
guardrail.score	Confidence score (set when the function returns a float value)

---

Pass/Fail Logic

The decorator determines the guardrail result based on the function's behavior:

Function Behavior	guardrail.result	guardrail.score
Returns normally (any type except float)	"pass"	Not set
Returns a float value	"pass"	Set to the returned float
Raises an exception	"fail"	Not set

Returning a score

If your guardrail computes a confidence score (e.g., PII detection probability, toxicity score), return it as a float. The score is recorded as guardrail.score and the result is still "pass". Raise an exception to signal "fail".

---

Examples

PII detection guardrail (pass case)

PythonTypeScript

from coalex.ext.guardrail import guardrail_span

@guardrail_span(guardrail_name="pii_detection")
def check_pii(text: str) -> float:
    """Check text for PII. Returns confidence that no PII is present."""
    pii_score = pii_detector.scan(text)
    if pii_score > 0.8:
        raise ValueError(f"PII detected with confidence {pii_score}")
    return 1.0 - pii_score  # confidence that text is clean

# Pass case: no PII detected
result = check_pii("The patient is doing well.")
# Span emitted with:
#   openinference.span.kind = "GUARDRAIL"
#   guardrail.name = "pii_detection"
#   guardrail.result = "pass"
#   guardrail.score = 0.95

import { guardrailSpan } from "@coalex-ai/sdk/ext";

const checkPii = guardrailSpan(
    { guardrailName: "pii_detection" },
    async (text: string): Promise<number> => {
        const piiScore = await piiDetector.scan(text);
        if (piiScore > 0.8) {
            throw new Error(`PII detected with confidence ${piiScore}`);
        }
        return 1.0 - piiScore; // confidence that text is clean
    },
);

// Pass case: no PII detected
const result = await checkPii("The patient is doing well.");
// Span emitted with:
//   openinference.span.kind = "GUARDRAIL"
//   guardrail.name = "pii_detection"
//   guardrail.result = "pass"
//   guardrail.score = 0.95

PII detection guardrail (fail case)

PythonTypeScript

try:
    result = check_pii("Patient SSN is 123-45-6789")
except ValueError:
    print("PII detected -- blocking output")
# Span emitted with:
#   openinference.span.kind = "GUARDRAIL"
#   guardrail.name = "pii_detection"
#   guardrail.result = "fail"

try {
    await checkPii("Patient SSN is 123-45-6789");
} catch (e) {
    console.log("PII detected -- blocking output");
}
// Span emitted with:
//   openinference.span.kind = "GUARDRAIL"
//   guardrail.name = "pii_detection"
//   guardrail.result = "fail"

Toxicity guardrail

PythonTypeScript

@guardrail_span(guardrail_name="toxicity_check")
def check_toxicity(text: str) -> float:
    """Check text for toxic content. Raises if toxic."""
    score = toxicity_model.predict(text)
    if score > 0.7:
        raise ValueError(f"Toxic content detected (score={score:.2f})")
    return 1.0 - score

# Safe content
check_toxicity("Thank you for your help!")
# guardrail.result = "pass", guardrail.score = 0.98

# Toxic content
try:
    check_toxicity("some toxic text here")
except ValueError:
    pass
# guardrail.result = "fail"

const checkToxicity = guardrailSpan(
    { guardrailName: "toxicity_check" },
    async (text: string): Promise<number> => {
        const score = await toxicityModel.predict(text);
        if (score > 0.7) {
            throw new Error(`Toxic content detected (score=${score.toFixed(2)})`);
        }
        return 1.0 - score;
    },
);

// Safe content
await checkToxicity("Thank you for your help!");
// guardrail.result = "pass", guardrail.score = 0.98

// Toxic content
try {
    await checkToxicity("some toxic text here");
} catch {
    // guardrail.result = "fail"
}

Factuality guardrail

PythonTypeScript

@guardrail_span(name="fact_check", guardrail_name="factuality_verification")
def verify_facts(claim: str, sources: list[str]) -> float:
    """Verify a claim against source documents. Returns factuality score."""
    result = fact_checker.verify(claim=claim, sources=sources)
    if result.score < 0.5:
        raise ValueError(f"Claim not supported by sources (score={result.score:.2f})")
    return result.score

score = verify_facts(
    claim="Metformin reduces HbA1c by 1-1.5%",
    sources=["Study A showed 1.2% reduction...", "Meta-analysis found 1.0-1.5%..."],
)
# guardrail.result = "pass", guardrail.score = 0.92

const verifyFacts = guardrailSpan(
    { name: "fact_check", guardrailName: "factuality_verification" },
    async (claim: string, sources: string[]): Promise<number> => {
        const result = await factChecker.verify({ claim, sources });
        if (result.score < 0.5) {
            throw new Error(`Claim not supported by sources (score=${result.score.toFixed(2)})`);
        }
        return result.score;
    },
);

const score = await verifyFacts(
    "Metformin reduces HbA1c by 1-1.5%",
    ["Study A showed 1.2% reduction...", "Meta-analysis found 1.0-1.5%..."],
);
// guardrail.result = "pass", guardrail.score = 0.92

Simple boolean guardrail

You do not need to return a score. Returning any non-float/non-number value (or undefined) still records "pass":

PythonTypeScript

@guardrail_span(guardrail_name="length_check")
def check_response_length(text: str, max_length: int = 5000) -> bool:
    """Ensure response is within length limits."""
    if len(text) > max_length:
        raise ValueError(f"Response too long: {len(text)} > {max_length}")
    return True  # guardrail.result = "pass", guardrail.score is NOT set (not a float)

const checkResponseLength = guardrailSpan(
    { guardrailName: "length_check" },
    async (text: string, maxLength = 5000): Promise<boolean> => {
        if (text.length > maxLength) {
            throw new Error(`Response too long: ${text.length} > ${maxLength}`);
        }
        return true; // guardrail.result = "pass", guardrail.score is NOT set (not a number)
    },
);

Async guardrail

PythonTypeScript

@guardrail_span(guardrail_name="content_policy")
async def check_content_policy(text: str) -> float:
    result = await moderation_api.check(text)
    if result.flagged:
        raise ValueError(f"Content policy violation: {result.categories}")
    return result.confidence

const checkContentPolicy = guardrailSpan(
    { guardrailName: "content_policy" },
    async (text: string): Promise<number> => {
        const result = await moderationApi.check(text);
        if (result.flagged) {
            throw new Error(`Content policy violation: ${result.categories}`);
        }
        return result.confidence;
    },
);

---

Full Pipeline Example

PythonTypeScript

import coalex
from coalex.ext.guardrail import guardrail_span
from openai import OpenAI

coalex.register(api_key="your-key")
coalex.auto_instrument()

client = OpenAI()

@guardrail_span(guardrail_name="input_validation")
def validate_input(text: str) -> float:
    if len(text) < 5:
        raise ValueError("Input too short")
    if len(text) > 10000:
        raise ValueError("Input too long")
    return 1.0

@guardrail_span(guardrail_name="pii_output_scan")
def scan_output_pii(text: str) -> float:
    import re
    patterns = {
        "ssn": r"\b\d{3}-\d{2}-\d{4}\b",
        "email": r"\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b",
        "phone": r"\b\d{3}[-.]?\d{3}[-.]?\d{4}\b",
    }
    for name, pattern in patterns.items():
        if re.search(pattern, text):
            raise ValueError(f"PII detected: {name}")
    return 1.0

@guardrail_span(guardrail_name="hallucination_check")
def check_hallucination(output: str, context: str) -> float:
    """Check if the output is grounded in the provided context."""
    result = nli_model.predict(premise=context, hypothesis=output)
    if result.label == "contradiction":
        raise ValueError(f"Hallucination detected (entailment={result.score:.2f})")
    return result.score

with coalex.coalex_context(agent_id="safe-agent", request_id="req-001"):
    user_input = "What medication is recommended for my condition?"

    # Pre-generation guardrails
    validate_input(user_input)

    # Generate response
    context = "Based on clinical guidelines, ibuprofen 400mg is recommended."
    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[
            {"role": "system", "content": f"Context: {context}"},
            {"role": "user", "content": user_input},
        ],
    )
    answer = response.choices[0].message.content

    # Post-generation guardrails
    scan_output_pii(answer)
    check_hallucination(output=answer, context=context)

    # Trace:
    #   coalex.invocation
    #     ├── validate_input (GUARDRAIL) - result=pass, score=1.0
    #     ├── openai.chat.completions (LLM)
    #     ├── scan_output_pii (GUARDRAIL) - result=pass, score=1.0
    #     └── check_hallucination (GUARDRAIL) - result=pass, score=0.91

import { register, coalexContext, autoInstrument } from "@coalex-ai/sdk";
import { guardrailSpan } from "@coalex-ai/sdk/ext";
import OpenAI from "openai";

register({ apiKey: "your-key" });
autoInstrument();

const client = new OpenAI();

const validateInput = guardrailSpan(
    { guardrailName: "input_validation" },
    async (text: string): Promise<number> => {
        if (text.length < 5) throw new Error("Input too short");
        if (text.length > 10000) throw new Error("Input too long");
        return 1.0;
    },
);

const scanOutputPii = guardrailSpan(
    { guardrailName: "pii_output_scan" },
    async (text: string): Promise<number> => {
        const patterns: Record<string, RegExp> = {
            ssn: /\b\d{3}-\d{2}-\d{4}\b/,
            email: /\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b/,
            phone: /\b\d{3}[-.]?\d{3}[-.]?\d{4}\b/,
        };
        for (const [name, pattern] of Object.entries(patterns)) {
            if (pattern.test(text)) throw new Error(`PII detected: ${name}`);
        }
        return 1.0;
    },
);

const checkHallucination = guardrailSpan(
    { guardrailName: "hallucination_check" },
    async (output: string, context: string): Promise<number> => {
        const result = await nliModel.predict({ premise: context, hypothesis: output });
        if (result.label === "contradiction") {
            throw new Error(`Hallucination detected (entailment=${result.score.toFixed(2)})`);
        }
        return result.score;
    },
);

await coalexContext({ agentId: "safe-agent", requestId: "req-001" }, async () => {
    const userInput = "What medication is recommended for my condition?";

    // Pre-generation guardrails
    await validateInput(userInput);

    // Generate response
    const context = "Based on clinical guidelines, ibuprofen 400mg is recommended.";
    const response = await client.chat.completions.create({
        model: "gpt-4o",
        messages: [
            { role: "system", content: `Context: ${context}` },
            { role: "user", content: userInput },
        ],
    });
    const answer = response.choices[0].message.content!;

    // Post-generation guardrails
    await scanOutputPii(answer);
    await checkHallucination(answer, context);

    // Trace:
    //   coalex.invocation
    //     ├── validate_input (GUARDRAIL) - result=pass, score=1.0
    //     ├── openai.chat.completions (LLM)
    //     ├── scan_output_pii (GUARDRAIL) - result=pass, score=1.0
    //     └── check_hallucination (GUARDRAIL) - result=pass, score=0.91
});

---

API Reference

coalex.ext.guardrail

Guardrail span decorator.

Functions

guardrail_span

guardrail_span(
    name: str | None = None,
    guardrail_name: str | None = None,
) -> Callable[[F], F]

Decorator that wraps a guardrail function with an OpenInference GUARDRAIL span.

Supports both sync and async functions. The span captures: - openinference.span.kind = GUARDRAIL - guardrail.name — from guardrail_name or the function name - guardrail.result — "pass" if no exception raised, "fail" otherwise - guardrail.score — set this via the return value if it is a float

Parameters:

Name	Type	Description	Default
name	str | None	Span name. Defaults to the function name.	None
guardrail_name	str | None	Logical guardrail name recorded as an attribute.	None

Source code in coalex/ext/guardrail.py

def guardrail_span(
    name: str | None = None,
    guardrail_name: str | None = None,
) -> Callable[[F], F]:
    """Decorator that wraps a guardrail function with an OpenInference GUARDRAIL span.

    Supports both sync and async functions. The span captures:
    - ``openinference.span.kind`` = ``GUARDRAIL``
    - ``guardrail.name`` — from ``guardrail_name`` or the function name
    - ``guardrail.result`` — ``"pass"`` if no exception raised, ``"fail"`` otherwise
    - ``guardrail.score`` — set this via the return value if it is a ``float``

    Args:
        name: Span name. Defaults to the function name.
        guardrail_name: Logical guardrail name recorded as an attribute.
    """

    def decorator(fn: F) -> F:
        span_name = name or fn.__name__
        logical_name = guardrail_name or span_name

        if asyncio.iscoroutinefunction(fn):

            @functools.wraps(fn)
            async def async_wrapper(*args: Any, **kwargs: Any) -> Any:
                tracer = _get_tracer()
                parent_ctx = _get_parent_context()
                with tracer.start_as_current_span(span_name, context=parent_ctx) as span:
                    span.set_attribute("openinference.span.kind", "GUARDRAIL")
                    span.set_attribute("guardrail.name", logical_name)
                    try:
                        result = await fn(*args, **kwargs)
                        span.set_attribute("guardrail.result", "pass")
                        if isinstance(result, float):
                            span.set_attribute("guardrail.score", result)
                        span.set_status(Status(StatusCode.OK))
                        return result
                    except Exception:
                        span.set_attribute("guardrail.result", "fail")
                        raise

            return async_wrapper  # type: ignore[return-value]

        @functools.wraps(fn)
        def wrapper(*args: Any, **kwargs: Any) -> Any:
            tracer = _get_tracer()
            parent_ctx = _get_parent_context()
            with tracer.start_as_current_span(span_name, context=parent_ctx) as span:
                span.set_attribute("openinference.span.kind", "GUARDRAIL")
                span.set_attribute("guardrail.name", logical_name)
                try:
                    result = fn(*args, **kwargs)
                    span.set_attribute("guardrail.result", "pass")
                    if isinstance(result, float):
                        span.set_attribute("guardrail.score", result)
                    span.set_status(Status(StatusCode.OK))
                    return result
                except Exception:
                    span.set_attribute("guardrail.result", "fail")
                    raise

        return wrapper  # type: ignore[return-value]

    return decorator