openai-omega

OpenAI Integration with OmegaEngine Governance

Documentation • Playground • Issues

✨ Features

• 🤖 Drop-in Replacement — Same API as official OpenAI SDK
• 🛡️ Pre-flight Validation — Block unsafe prompts before sending
• 📊 Response Governance — Filter outputs against policies
• 📈 Full Audit Trail — Log every GPT-4/GPT-5 interaction
• ⚡ Streaming Support — Governed streaming responses
• 🔧 Function Call Validation — Validate tool calls before execution
• 🎯 Assistants API — Govern OpenAI Assistants

📦 Installation

Python

pip install openai-omega

🚀 Quick Start (Python)

from openai_omega import OpenAI

# Drop-in replacement for OpenAI client
client = OpenAI(
    openai_api_key="your-openai-key",
    omega_api_key="your-omega-key",
    policy_id="gpt_policy",
)

# Same API, but governed
response = client.chat.completions.create(
    model="gpt-4-turbo",
    messages=[
        {"role": "user", "content": "Help me with marketing"}
    ]
)

print(response.choices[0].message.content)

# Access governance metadata
print(response.omega_decision)      # "ALLOW" | "BLOCK" | "ESCALATE"
print(response.omega_audit_id)      # "aud_abc123"
print(response.omega_risk_score)    # 15

🛡️ Pre-flight Validation

Block unsafe prompts before they reach OpenAI:

client = OpenAI(
    openai_api_key="...",
    omega_api_key="...",
    preflight_check=True,
    block_on_deny=True,
)

try:
    response = client.chat.completions.create(
        model="gpt-4-turbo",
        messages=[{"role": "user", "content": "...jailbreak attempt..."}]
    )
except OmegaBlockedError as e:
    print(f"Blocked: {e.reason}")
    print(f"Risk score: {e.risk_score}")

🔧 Function Call Governance

Validate function/tool calls before execution:

response = client.chat.completions.create(
    model="gpt-4-turbo",
    messages=[...],
    tools=[...],
    omega_validate_tools=True,  # Validate tool calls
)

for tool_call in response.choices[0].message.tool_calls:
    if tool_call.omega_approved:
        execute_tool(tool_call)
    else:
        print(f"Tool blocked: {tool_call.omega_reason}")

📡 Governed Streaming

with client.chat.completions.create(
    model="gpt-4-turbo",
    messages=[{"role": "user", "content": "Write a story"}],
    stream=True,
) as stream:
    for chunk in stream:
        print(chunk.choices[0].delta.content, end="")

print(f"\nDecision: {stream.omega_decision}")

🤖 Assistants API Support

# Governed Assistants
assistant = client.beta.assistants.create(
    name="Analyst",
    model="gpt-4-turbo",
    omega_policy_id="assistant_policy",
)

# All threads are governed
thread = client.beta.threads.create()
message = client.beta.threads.messages.create(
    thread_id=thread.id,
    role="user",
    content="Analyze this data..."
)

run = client.beta.threads.runs.create(
    thread_id=thread.id,
    assistant_id=assistant.id,
)

# Access governance metadata
print(run.omega_audit_id)

📊 What Gets Logged

📄 License

Licensed under the Apache License 2.0.

Built with ❤️ by the OmegaEngine team