TruthVouch Python SDK

Drop-in LLM governance for Python. Route OpenAI, Anthropic, and Google AI calls through the TruthVouch Governance Gateway for automatic policy enforcement, PII detection, and hallucination monitoring.

Installation

pip install truthvouch

Optional extras:

pip install "truthvouch[telemetry]"   # OpenTelemetry + Prometheus
pip install "truthvouch[adapters]"    # LangGraph + CrewAI integrations
pip install "truthvouch[all]"         # Everything

Quick Start — Gateway Proxy

from truthvouch import TruthVouch

async with TruthVouch(gateway_url="https://gateway.truthvouch.com", api_key="tv_live_...") as client:
    # OpenAI drop-in
    resp = await client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": "What is quantum computing?"}],
    )
    print(resp.choices[0].message.content)
    print(resp.governance.verdict)       # "allowed" | "blocked" | "flagged"
    print(resp.governance.pii_detected)  # False

Provider Facades

# Anthropic
resp = await client.anthropic.messages.create(
    model="claude-3-5-sonnet-20241022",
    messages=[{"role": "user", "content": "Hello"}],
    max_tokens=512,
)
print(resp.text)

# Google AI
resp = await client.google.generate_content(
    model="gemini-1.5-pro",
    contents=[{"role": "user", "parts": [{"text": "Hello"}]}],
)
print(resp.text)

Streaming

async for chunk in await client.chat.completions.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": "Tell me a story"}],
    stream=True,
):
    print(chunk.content, end="", flush=True)
    if chunk.done:
        print(f"\nVerdict: {chunk.governance_report.verdict}")

Batch Scanning

job = await client.batch.submit(source_url="s3://my-bucket/prompts.jsonl", format="jsonl")
status = await client.batch.get_status(job.id)
print(status.status)  # "completed"

Governance Guards

For fine-grained input/output evaluation with local and remote guards:

from truthvouch.governance import GovernanceClient

client = GovernanceClient(api_key="vt_live_...")

result = client.evaluate_input("My SSN is 123-45-6789", model="gpt-4o")
if result.blocked:
    print("Blocked:", result.block_reasons)

# Async usage
from truthvouch.governance import AsyncGovernanceClient

async with AsyncGovernanceClient(api_key="vt_live_...") as client:
    result = await client.evaluate_input("Hello!", model="gpt-4o")

Built-in Guards

Framework Adapters

LangGraph / LangChain:

from truthvouch.governance.adapters.langgraph import TruthVouchCallbackHandler

handler = TruthVouchCallbackHandler(client, model="gpt-4o")
result = chain.invoke(inputs, config={"callbacks": [handler]})

CrewAI:

from truthvouch.governance.adapters.crewai import TruthVouchMiddleware

middleware = TruthVouchMiddleware(client)
llm = middleware.wrap_llm(ChatOpenAI(model="gpt-4o"))
agent = Agent(llm=llm, role="Analyst", goal="...", backstory="...")

Error Handling

from truthvouch.exceptions import (
    PolicyBlockedError,
    AuthenticationError,
    QuotaExceededError,
    GatewayUnreachableError,
)

try:
    resp = await client.chat.completions.create(model="gpt-4o", messages=[...])
except PolicyBlockedError as e:
    print(f"Blocked: {e.governance_report}")
except AuthenticationError:
    print("Invalid API key")
except QuotaExceededError as e:
    print(f"Rate limited, retry after {e.retry_after_seconds}s")
except GatewayUnreachableError:
    print("Gateway down — call went direct to LLM (fail-open mode)")

Configuration

from truthvouch import TruthVouch

client = TruthVouch(
    gateway_url="https://gateway.truthvouch.com",
    api_key="tv_live_...",
    fail_mode="open",         # "open" (bypass) or "closed" (raise error)
    timeout=30.0,             # HTTP timeout in seconds
    max_retries=3,            # Retry attempts with exponential backoff
    circuit_breaker_threshold=5,          # Failures before circuit opens
    circuit_breaker_recovery_seconds=60,  # Seconds before recovery probe
    verify_ssl=True,          # TLS certificate verification
)

License

Apache-2.0