sensoit 1.0.1

Sensoit Python SDK - AI Safety & Observability Platform

Sensoit Python SDK

AI Safety & Observability Platform - Python SDK

The Sensoit SDK provides automatic tracing, guardrails, budget management, and safety features for your LLM applications. Every traced call automatically gets:

• Guardrail checks - PII detection, toxicity filtering, jailbreak prevention
• Budget enforcement - Token limits, cost limits, time limits with auto-abort
• Contradiction detection - Detect conflicting outputs across agent steps
• Auto-rollback - Automatic rollback on safety violations

Performance target: <2ms overhead per call

Installation

# Core SDK
pip install sensoit

# With OpenAI
pip install sensoit[openai]

# With Anthropic
pip install sensoit[anthropic]

# With LangChain
pip install sensoit[langchain]

# With CrewAI
pip install sensoit[crewai]

# All integrations
pip install sensoit[all]

Quick Start

1. Initialize

import sensoit

# Initialize once at startup
sensoit.init(api_key="gf_live_xxx")

# Or use environment variable
# export SENSOIT_API_KEY=gf_live_xxx
sensoit.init()

2. Wrap Your LLM Client

from openai import OpenAI
import sensoit

sensoit.init(api_key="gf_live_xxx")

# Wrap the client - all calls are now traced with safety
client = sensoit.wrap_openai(OpenAI())

response = client.chat.completions.create(
    model="gpt-4",
    messages=[{"role": "user", "content": "Hello!"}]
)

3. Use Sessions for Agent Workflows

import sensoit
from openai import OpenAI

sensoit.init(api_key="gf_live_xxx")
client = sensoit.wrap_openai(OpenAI())

# Session provides budget enforcement and safety tracking
with sensoit.session(
    "my-agent",
    budget={"max_tokens": 5000, "max_cost": 0.50}
) as session:
    response = client.chat.completions.create(
        model="gpt-4",
        messages=[{"role": "user", "content": "Research AI safety"}]
    )

    print(f"Tokens used: {session.total_tokens}")
    print(f"Cost: ${session.total_cost:.4f}")

Provider Integrations

OpenAI

from openai import OpenAI
import sensoit

sensoit.init()
client = sensoit.wrap_openai(OpenAI())

# Streaming works too
for chunk in client.chat.completions.create(
    model="gpt-4",
    messages=[{"role": "user", "content": "Hello"}],
    stream=True
):
    print(chunk.choices[0].delta.content, end="")

Anthropic

from anthropic import Anthropic
import sensoit

sensoit.init()
client = sensoit.wrap_anthropic(Anthropic())

response = client.messages.create(
    model="claude-3-sonnet-20240229",
    max_tokens=1024,
    messages=[{"role": "user", "content": "Hello!"}]
)

Groq

from groq import Groq
import sensoit

sensoit.init()
client = sensoit.wrap_groq(Groq())

response = client.chat.completions.create(
    model="llama3-8b-8192",
    messages=[{"role": "user", "content": "Hello!"}]
)

Mistral

from mistralai import Mistral
import sensoit

sensoit.init()
client = sensoit.wrap_mistral(Mistral())

response = client.chat.complete(
    model="mistral-small-latest",
    messages=[{"role": "user", "content": "Hello!"}]
)

LiteLLM (Universal)

import litellm
import sensoit

sensoit.init()
sensoit.wrap_litellm()  # Patches globally

# Now use any model
response = litellm.completion(
    model="gpt-4",
    messages=[{"role": "user", "content": "Hello"}]
)

Google Generative AI

import google.generativeai as genai
import sensoit

sensoit.init()

model = genai.GenerativeModel("gemini-pro")
model = sensoit.wrap_google(model)

response = model.generate_content("Hello!")

Framework Integrations

LangChain

from langchain_openai import ChatOpenAI
import sensoit

sensoit.init()
handler = sensoit.LangChainCallbackHandler()

llm = ChatOpenAI(
    model="gpt-4",
    callbacks=[handler]
)

result = llm.invoke("Hello!")

LlamaIndex

from llama_index.core import Settings, VectorStoreIndex
from llama_index.core.callbacks import CallbackManager
import sensoit

sensoit.init()
handler = sensoit.LlamaIndexCallbackHandler()

Settings.callback_manager = CallbackManager([handler])

# All LlamaIndex operations are now traced
index = VectorStoreIndex.from_documents(documents)
response = index.query("What is AI safety?")

CrewAI

from crewai import Crew, Agent, Task
import sensoit

sensoit.init()

agent = Agent(
    role="Researcher",
    goal="Research topics thoroughly",
    backstory="You are a skilled researcher.",
)

task = Task(
    description="Research AI safety best practices",
    agent=agent,
)

crew = Crew(
    agents=[agent],
    tasks=[task],
)

# Add callback for tracing
result = crew.kickoff(callbacks=[sensoit.CrewAICallback()])

LangGraph

from langgraph.graph import StateGraph, END
import sensoit

sensoit.init()
tracer = sensoit.LangGraphTracer()

# Define your graph
graph = StateGraph(State)
graph.add_node("process", process_fn)
graph.add_edge("process", END)

app = graph.compile()

# Run with tracer
result = app.invoke(
    {"input": "Hello"},
    config={"callbacks": [tracer]}
)

Decorators

@sensoit.trace

Trace any function:

import sensoit

sensoit.init()

@sensoit.trace("search_documents", type="RETRIEVAL")
def search_documents(query: str) -> list:
    # Your search logic
    return results

@sensoit.trace("process_query", type="TOOL")
async def process_query(query: str) -> str:
    # Your async logic
    return result

@sensoit.agent

Wrap a function as an agent with automatic session management:

import sensoit

sensoit.init()

@sensoit.agent("research-agent", budget={"max_tokens": 5000, "max_cost": 0.50})
def run_research(query: str) -> str:
    # All LLM calls within this function are tracked
    # Budget is automatically enforced
    return research_result

Budget Management

Set limits to prevent runaway costs:

import sensoit
from sensoit import BudgetExceededError

sensoit.init()

try:
    with sensoit.session(
        "my-agent",
        budget={
            "max_tokens": 1000,    # Token limit
            "max_cost": 0.10,      # Dollar limit
            "max_seconds": 30,     # Time limit
        },
        auto_abort=True  # Raise exception on limit exceeded
    ) as session:
        # Your agent logic
        response = client.chat.completions.create(...)

except BudgetExceededError as e:
    print(f"Budget exceeded: {e.limit_type} - used {e.current_value}, limit {e.limit_value}")

Feedback Collection

Collect user feedback on LLM outputs:

import sensoit

sensoit.init()

# After getting a response
span_id = "span_xxx"  # From your tracing

# Binary feedback
sensoit.feedback.thumbs_up(span_id=span_id)
sensoit.feedback.thumbs_down(span_id=span_id, comment="Incorrect answer")

# Rating
sensoit.feedback.rating(4.5, span_id=span_id, max_score=5.0)

# Correction
sensoit.feedback.correction(
    "The correct answer is...",
    span_id=span_id
)

Evaluations

Run evaluations on your prompts:

import sensoit

sensoit.init()

evaluator = sensoit.Evaluator("my-eval")

# Add test cases
evaluator.add_case("test-1", {"query": "What is 2+2?"}, expected="4")
evaluator.add_case("test-2", {"query": "Capital of France?"}, expected="Paris")

# Run evaluation
results = evaluator.run(
    my_llm_function,
    scorer="contains",  # or "exact" or custom function
    max_workers=5
)

print(f"Passed: {results.passed}/{results.total}")
print(f"Avg latency: {results.avg_latency_ms:.1f}ms")

Manual Tracing

For custom operations:

import sensoit

sensoit.init()

with sensoit.span("custom_operation", type="TOOL") as span:
    span.set_input({"query": "some query"})

    # Your logic
    result = do_something()

    span.set_output(result)

Configuration

Environment Variables

# Required
SENSOIT_API_KEY=gf_live_xxx

# Optional
SENSOIT_BASE_URL=https://api.sensoit.io  # Custom API endpoint
SENSOIT_DEBUG=true                        # Enable debug logging

Init Options

sensoit.init(
    api_key="gf_live_xxx",           # API key
    base_url="https://api.sensoit.io", # API endpoint
    flush_interval=5.0,               # Batch flush interval (seconds)
    max_batch_size=100,               # Max spans per batch
    guardrails_enabled=True,          # Enable guardrail checks
    debug=False,                      # Debug logging
)

Graceful Shutdown

The SDK automatically flushes on exit, but you can force flush:

# Force flush all pending spans
sensoit.force_flush()

# Manual shutdown
sensoit.shutdown()

API Reference

Core Functions

Function	Description
sensoit.init()	Initialize the SDK
sensoit.shutdown()	Shutdown and flush
sensoit.force_flush()	Force flush pending spans
sensoit.session()	Create a traced session
sensoit.span()	Create a custom span
sensoit.get_session()	Get current session
sensoit.get_tracer()	Get tracer instance

Provider Wrappers

Function	Description
sensoit.wrap_openai()	Wrap OpenAI client
sensoit.wrap_anthropic()	Wrap Anthropic client
sensoit.wrap_groq()	Wrap Groq client
sensoit.wrap_mistral()	Wrap Mistral client
sensoit.wrap_litellm()	Patch LiteLLM globally
sensoit.wrap_google()	Wrap Google AI client

Framework Integrations

Function	Description
sensoit.LangChainCallbackHandler()	LangChain callback
sensoit.LlamaIndexCallbackHandler()	LlamaIndex callback
sensoit.CrewAICallback()	CrewAI callback
sensoit.LangGraphTracer()	LangGraph tracer

Decorators

Decorator	Description
@sensoit.trace()	Trace a function
@sensoit.agent()	Wrap function as agent with session

Feedback

Function	Description
sensoit.feedback.thumbs_up()	Submit positive feedback
sensoit.feedback.thumbs_down()	Submit negative feedback
sensoit.feedback.rating()	Submit numeric rating
sensoit.feedback.correction()	Submit correction

License

MIT