Python SDK

Core Python SDK supports server-side data ingestion to Align

Installation

Install the Python SDK with pip:

pip install alignai

Usage

The Python SDK must be authenticated with a valid API key. If you haven't generated an API key yet, visit Align.

Initialize the SDK

To send events, initialize the SDK using your API key and Project ID:

from alignai import AlignAI

sdk = AlignAI(
    project_id="PROJECT_ID",
    api_key="API_KEY"
)

Identify User

Use the identify_user method to record detailed user information.

The most recommended use cases of this method are:

When a user first signs up for your service.
When a user updates their profile information such as email address or display name.
When open_session event is emitted.
- While the identify_user method can be employed during sign-up or profile updates, an alternate strategy is to initially avoid the identify_user event altogether. Given that the open_session event inherently demands a user_id, you can solely focus on this event. By doing so, the user's identification process will be seamlessly managed.

The user_id is the distinct identifier given to every user in your service. You can employ your existing user ID or create a new one using methods similar to uuid.

On the other hand, display_name is the user's display name shown on dashboard.

user_id (string, required): Unique user ID used internally. This ID will be not displayed on dashboard. Max 64 chars allowed.
display_name (string): User display name shown on dashboard. It's adviced to use the user's nickname for this purpose. If a nickname isn't available, alternatives like the email address can be used. This field does not have to be unique. Defaults to None. Max 64 chars allowed.
email (string): User email address. Max 256 chars allowed.
ip (string): User IPv4 address. Provide either ip or country code for user location. If both are given, country code overrides ip.
country_code (string): User country code in ISO Alpha-2. Provide either ip or country code for user location. If both are given, country code overrides ip.
create_time (datetime): User creation time.
custom_properties (dict[str, str]): Custom Properties(Metadata) of user. Key should match pattern ^[a-zA-Z][a-zA-Z0-9_]{0,63}$. Max 256 chars allowed for value. Up to 10 properties are allowed. (This is merely a global limit. Please verify your project limit on your settings page.)

sdk.identify_user(
    user_id="USER_ID",
    display_name="USER_DISPLAY_NAME",
    email="USER_EMAIL",
    country_code="ALPHA2_COUNTRY_CODE",
    ip="USER_IP",
    create_time="USER_CREATE_TIME",
)

Open Session

Use the open_session method to record the initiation of a session.

session_id (string, required): Unique session ID used internally. This ID will be not displayed on dashboard. Max 64 chars allowed.
user_id (string, required): User ID that the session belongs to. Max 64 chars allowed.
assistant_id (string): Assistant ID, if any. Max 64 chars allowed.
custom_properties (dict[str, str]): Custom Properties(Metadata) of session. Key should match pattern ^[a-zA-Z][a-zA-Z0-9_]{0,63}$. Max 256 chars allowed for value. Up to 10 properties are allowed. (This is merely a global limit. Please verify your project limit on your settings page.)

sdk.open_session(
    session_id="SESSION_ID",
    user_id="USER_ID",
    assistant_id="ASSISTANT_ID"
)

Create Message

Use the create_message method to record an individual message within a session.

session_id (string, required): Session ID the message belongs to. Max 64 chars allowed.
message_index (int, required): Message index used to sort messages in a chronological order within a session. Must be unique within a session and should be a positive integer.
role (string, required): "user" or "assistant".
content (string, required): Content of the message.
custom_properties (dict[str, str]): Custom Properties(Metadata) of message. Key should match pattern ^[a-zA-Z][a-zA-Z0-9_]{0,63}$. Max 256 chars allowed for value. Up to 10 properties are allowed. (This is merely a global limit. Please verify your project limit on your settings page.)

from alignai.constants import ROLE_USER

sdk.create_message(
    session_id="SESSION_ID",
    message_index=1,
    role=ROLE_USER,
    content="Hey, you there?"
)

Close Session

Use the close_session method to record the end of a session.

If the session is inactive for 30 minutes, close_session event will be automatically emitted.

session_id (string, required): Session ID. Max 64 chars allowed.

sdk.close_session(
    session_id="SESSION_ID"
)

Log a feedback for a message

Use the create_message_feedback method to record the feedback for a message.

session_id (string, required): Session ID the message belongs to. Max 64 chars allowed.
message_index (int, required): Message index used to sort messages in a chronological order within a session. Must be unique within a session and should be a positive integer.
feedback_type (string, required): "thumbs_up" or "thumbs_down". You can use pre-defined constants.

Log a feedback for a session

Use the create_session_feedback method to record the feedback for a message.

session_id (string, required): Session ID the message belongs to. Max 64 chars allowed.
feedback_type (string, required): "thumbs_up" or "thumbs_down". You can use pre-defined constants.

flush

The Python SDK utilizes batch processing for efficient event ingestion.

To instantly dispatch all events from the buffer, invoke the flush method.

timeout_seconds (int): After the timeout, SDK will stop flushing. Default to 5.

sdk.flush()

close

If you wish to terminate the SDK, call the close method. This method will first flush all events and then shut down the SDK.

For a graceful shutdown, it's advised to use this method.

timeout_seconds (int): After the timeout, SDK will stop flushing and the remaining data in the buffer will be lost. Default to 5.

sdk.close()

Example Usage

The code snippet below shows a straightforward demonstration of utilizing the Python SDK to send events to Align.

from alignai import AlignAI
from alignai.constants import ROLE_USER, ROLE_ASSISTANT
import openai

openai.api_key = os.environ.get("OPENAI_API_KEY")

# Initialize SDK
sdk = AlignAI(
    project_id="TEFSZ16",
    api_key="NIUaqth_TsuIQWox_TfD5sCEQ8PebffFjF76u7Vsl2o="
)

user_id = "d75d27cbff184b54917f7b921cd2afce"
# Identify a user
sdk.identify_user(
    user_id=user_id,
    email="sample-user@gmail.com",
    country_code="US",
)

# Open a session
session_id = uuid.uuid4().hex
assistant_id = "1d2bfd85babd4804aca1-656f5d91dfa"
sdk.open_session(
    session_id=session_id,
    user_id=user_id,
    assistant_id=assistant_id
)

# Send initial message
history = []
initial_message = "Hello, what can I do for u?"
history.append({"role": ROLE_ASSISTANT, "content": initial_message})
sdk.create_message(
    session_id=session_id,
    message_index=1,
    role=ROLE_ASSISTANT,
    content=initial_message
)

try:
    while True:
        # Get user input
        user_input = input("User: ")
        history.append({"role": ROLE_USER, "content": user_input})

        # Send user message
        sdk.create_message(
            session_id=session_id,
            message_index=len(history),
            role=ROLE_USER,
            content=user_input
        )

        # Get assistant response
        response = openai.ChatCompletion.create(
            model="gpt-3.5-turbo",
            messages=history
        )
        assistant_message = response.choices[0].message['content'].strip()
        history.append({"role": ROLE_ASSISTANT, "content": assistant_message})    

        # Send assistant message
        sdk.create_message(
            session_id=session_id,
            message_index=len(history),
            role=ROLE_ASSISTANT,
            content=assistant_message
        )
        print("Assistant:", assistant_message)
finally:
    # Close the session
    sdk.close_session(
        session_id=session_id
    )
    # Flush the unsent events close the SDK. 
    # If you don't close the SDK, you might lose remaining events in the buffer.
    sdk.close(timeout_seconds=5)

Last modified: 28 February 2025