Python SDK

The ZeebeeAI Python client library provides a simple and powerful way to integrate AI functionality into your Python applications, including multi-model support, agent orchestration, and advanced routing capabilities.

Installation

You can install the ZeebeeAI Python SDK using pip:

pip install zeebee-ai-client

AI Chat

The ZeebeeAI SDK provides a powerful chat interface for interacting with large language models through the chat() method of the ZeebeeClient class.

Function Signature

Basic Example

Here's a simple example of how to use the SDK for basic chat functionality:

import os
from zeebee_ai_client import ZeebeeClient

# Initialize the client
client = ZeebeeClient(
    api_key=os.environ.get("ZEEBEE_API_KEY"),  # Replace with your API key
)

# Send a simple message
response = client.chat("Hello, how can you help me today?")
print(response["message"])

# Or create a conversation for multi-turn interactions
conversation = client.create_conversation()
conversation_id = conversation["id"]
response = client.chat("Tell me about quantum computing", conversation_id=conversation_id)
print(response["message"])

# Continue the conversation
follow_up = client.chat("How is that related to cryptography?", conversation_id=conversation_id)
print(follow_up["message"])

Voice Chat

The ZeebeeAI SDK provides WebSocket-based voice chat functionality for real-time audio streaming interactions. This allows your application to send audio recordings and receive responses both as text and audio.

WebSocketVoiceChat Class

VoiceChatSession Class

Basic Example

Here's a simple example of how to use the voice chat functionality:

import os
import asyncio
import soundfile as sf  # pip install soundfile
import sounddevice as sd  # pip install sounddevice
from zeebee_ai_client import WebSocketVoiceChat

# Example callback functions with practical implementations
def on_transcript(text):
    """Called when your audio is transcribed to text"""
    print(f"\n🎤 You said: {text}")
    # You could update UI with transcription here
    
def on_response(text):
    """Called when AI responds with text"""
    print(f"\n🤖 AI response: {text}")
    # You could update UI with AI response here
    
def on_audio(audio_data):
    """Called when AI responds with audio"""
    # Save the audio to a file
    with open("ai_response.wav", "wb") as f:
        f.write(audio_data)
    
    # If you have soundfile/sounddevice installed, play the audio
    try:
        # Save temporarily and play
        with open("temp_response.wav", "wb") as f:
            f.write(audio_data)
        
        # Load and play audio
        data, samplerate = sf.read("temp_response.wav")
        sd.play(data, samplerate)
        sd.wait()  # Wait until audio is finished playing
        print(f"🔊 Playing audio response ({len(audio_data)} bytes)")
    except Exception as e:
        print(f"Could not play audio: {e}")
        print(f"Audio saved to ai_response.wav ({len(audio_data)} bytes)")
    
def on_error(error_message):
    """Called when an error occurs"""
    print(f"\n❌ Error: {error_message}")
    
def on_status(status):
    """Called when connection status changes"""
    status_symbols = {
        "connecting": "🔄",
        "connected": "🟢",
        "ready": "✅",
        "processing": "⏳",
        "disconnected": "🔴",
        "reconnecting": "🔁"
    }
    symbol = status_symbols.get(status, "ℹ️")
    print(f"{symbol} Status: {status}")

async def main():
    # Initialize the WebSocket voice chat client
    voice_chat = WebSocketVoiceChat()
    
    # Create a voice chat session with our callback functions
    session = voice_chat.create_session(
        model="gpt-4o",  # AI model to use
        on_transcript=on_transcript,
        on_response=on_response,
        on_audio=on_audio,
        on_error=on_error,
        on_status=on_status
    )
    
    # Connect to the WebSocket server
    connected = await session.connect()
    if not connected:
        print("Failed to connect to voice chat server")
        return
    
    # Send an audio file to the server
    audio_file_path = "recording.wav"  # Path to your WAV audio file
    
    print(f"\nSending audio file: {audio_file_path}")
    success = await session.send_audio(audio_file_path)
    
    if success:
        print("Audio sent successfully")
    else:
        print("Failed to send audio")
    
    # Wait for the interaction to complete (transcription + AI response)
    # Adjust this time based on your expected response time
    print("Waiting for response...")
    await asyncio.sleep(30)
    
    # Disconnect when done
    await session.disconnect()
    await voice_chat.close_all_sessions()
    print("\nSession ended")

# Run the async main function
if __name__ == "__main__":
    asyncio.run(main())

pip install soundfile sounddevice

Advanced Features

ZeebeeAI SDK offers several advanced features including Agent Orchestration, Autonomous Routing, and Dynamic Layout Engine.

Dynamic Layout Engine

The Dynamic Layout Engine allows applications to intelligently generate appropriate layouts for content based on message complexity, detected intents, and context. This helps create visually rich and contextually appropriate user interfaces for AI responses.

LayoutController Class

LayoutType Class

Example:

from zeebee_ai_client import ZeebeeClient, LayoutController, LayoutType
import os

# Initialize the SDK
client = ZeebeeClient(api_key=os.environ.get("ZEEBEE_API_KEY"))

# Initialize the layout controller for layout generation
layout = LayoutController(client)

# Generate a dynamic layout for tabular data
table_layout = layout.generate_layout(
    message="Create a table comparing Python, JavaScript, and Java programming languages",
    routing_result={
        "suggested_template": LayoutType.TABLE_LAYOUT,
        "content_analysis": {
            "contentTypes": ["table", "comparison"],
            "complexity": "medium",
            "formality": "formal"
        }
    }
)

print(f"Generated layout template: {table_layout['layout']['template']}")
print(f"Layout type: {table_layout['layout']['type']}")
print(f"Number of components: {len(table_layout['layout']['components'])}")
print(f"Content analysis: {table_layout['layout']['content_analysis']}")

# Accessing specific components in the layout
for component in table_layout['layout']['components']:
    print(f"Component type: {component['type']}")
    if component['type'] == 'table':
        print(f"Table options: {component['options']}")

Agent Orchestration

The Agent Orchestration feature allows you to create, configure, and execute specialized AI agents designed for specific tasks, and connect them into pipelines for multi-step workflows. This enables complex AI applications that require multiple processing steps or different types of expertise.

AgentController Class

AgentTypes Class

PipelineController Class

Example:

from zeebee_ai_client import ZeebeeClient, AgentController, AgentTypes, PipelineController
import os

# Initialize the SDK
client = ZeebeeClient(api_key=os.environ.get("ZEEBEE_API_KEY"))

# Initialize the agents controller
agents = AgentController(client)

# Initialize the pipeline controller
pipelines = PipelineController(client)

# Create a specialized research agent
research_agent = agents.create_agent(
    name="Research Agent",
    agent_type=AgentTypes.RETRIEVAL,
    configuration={
        "description": "Retrieves and analyzes research information",
        "system_prompt": "You are a specialized research assistant. Provide detailed, accurate information with sources."
    },
    description="A specialized assistant for research tasks",
    model_id="gpt-4o"
)

# Get the agent details
agent_id = research_agent["agent_id"]
agent_details = agents.get_agent(agent_id)

# Update the agent configuration
updated_agent = agents.update_agent(
    agent_id,
    update_data={
        "name": "Advanced Research Agent",
        "configuration": {
            "system_prompt": "You are an advanced research assistant specializing in scientific literature. Provide comprehensive analysis with peer-reviewed sources."
        }
    }
)

# Execute the agent directly
result = agents.execute_agent(
    agent_id=agent_id,
    input_data={"query": "Summarize recent developments in nuclear fusion technology"}
)

# Get all available agent types
agent_types = agents.get_agent_types()
print(f"Available agent types: {agent_types}")

# Create a summarization agent
summarize_agent = agents.create_agent(
    name="Summarize Agent",
    agent_type=AgentTypes.SUMMARIZATION,
    configuration={
        "system_prompt": "You are a summarization expert. Create concise, accurate summaries of complex information."
    },
    description="An agent that summarizes complex information",
    model_id="gpt-4o"
)

# Create a pipeline of multiple agents
pipeline = pipelines.create_pipeline(
    name="Research and Summarize",
    stages=[
        {
            "agent_id": agent_id,
            "name": "Research Stage",
            "input_mapping": {"query": "$.input.research_topic"},
            "output_mapping": {"research_results": "$.output.result"}
        },
        {
            "agent_id": summarize_agent["agent_id"],
            "name": "Summarize Stage",
            "input_mapping": {"text": "$.stages.Research Stage.research_results"},
            "output_mapping": {"summary": "$.output.result"}
        }
    ],
    description="Pipeline that researches a topic and summarizes the findings"
)

# Execute the pipeline
pipeline_id = pipeline["pipeline_id"]
pipeline_result = pipelines.execute_pipeline(
    pipeline_id=pipeline_id,
    input_data={"research_topic": "Recent breakthroughs in quantum computing"}
)

# Get all pipelines
all_pipelines = pipelines.get_pipelines()
print(f"Number of pipelines: {len(all_pipelines['pipelines'])}")

# Get specific pipeline details
pipeline_details = pipelines.get_pipeline(pipeline_id)

# When done with the agent and pipeline, you can delete them
# agents.delete_agent(agent_id)
# agents.delete_agent(summarize_agent["agent_id"])
# pipelines.delete_pipeline(pipeline_id)

Autonomous Routing

The Autonomous Routing feature intelligently routes user messages to the appropriate agent, pipeline, or language model based on message content analysis. This enables your application to automatically direct user requests to specialized handling components without requiring explicit user selection.

RoutingController Class

IntentCategory Class

Example:

from zeebee_ai_client import ZeebeeClient, RoutingController
import os

# Initialize the SDK
client = ZeebeeClient(api_key=os.environ.get("ZEEBEE_API_KEY"))

# Initialize the routing controller
routing = RoutingController(client)

# Route a message to the appropriate agent, pipeline, or model
routing_result = routing.route_message(
    message="Can you analyze this data and create a visualization?"
)

print(f"Routing to: {routing_result['route_to']} ({routing_result['route_type']})")
print(f"Confidence: {routing_result['confidence']}")
print(f"Intent information: {routing_result['intent']}")
print(f"Reasoning: {routing_result['reasoning']}")