Python SDK Usage

# A simple chatbot server that integrates with the Mindlytics service

from fastapi import FastAPI, Request
from pydantic import BaseModel
from fastapi.responses import StreamingResponse
import asyncio
import os
from openai import OpenAI
from dotenv import load_dotenv
from mlsdk import Client as MLClient

load_dotenv()
app = FastAPI()
openai = OpenAI(api_key=os.environ.get("OPENAI_API_KEY"))

@app.middleware("http")
async def add_session_to_request(request: Request, call_next):
    # Get client-side ids from headers
    session_id = request.headers.get("x-session-id")
    conversation_id = request.headers.get("x-conversation-id")
    user_id = request.headers.get("x-user-id")
    device_id = request.headers.get("x-device-id")

    # Instanciate a client object with your organization api key and project id
    ml = MLClient(
        api_key=os.environ.get("MLSDK_API_KEY",
        project_id=os.environ.get("MLSDK_PROJECT_ID",
    )

    # session_id is required.  It should be a unique uuid representing a unique user session.
    # conversation_id is optional, but required if sending conversation-related events.
    # One of user_id or device_id is required.
    request.mlsession = ml.create_session(
        session_id=session_id,
        conversation_id=conversation_id,
        id=user_id,
        device_id=device_id,
    )

    try:
        response = await call_next(request)
    finally:
        # Flush the mindlytics event queue to ensure all messages are sent
        await request.mlsession.flush()

    return response

# Simple streamer function to stream tokens back to the client.
async def openai_streamer(mlsession, question: str):
    assistant = ""
    stream = client.chat.completions.create(
        model="gpt-4",
        messages=[{"role": "user", "content": question}],
        stream=True,
    )
    for chunk in stream:
        content = chunk.choices[0].delta.content or ""
        assistant.append(content)  # Capture the entire response ...
        yield content

    # Send the complete "turn" to the Mindlytics service
    await mlsession.track_conversation_turn(
        user: question,
        assistant: assistant
    )

class AskRequest(BaseModel):
    question: str

# Endpoint for asking a question of the assistant
@app.post("/ask-assistant")
async def ask_assistant(ask_request: AskRequest):
    return StreamingResponse(openai_streamer(ask_request.mlsession, ask_request.question), media_type="text/event-stream")

# Call this when the session/conversation is finished.
@app.post("end-session")
async def end-session(request: Request):
    await request.mlsession.end_session()
    return {"status": "ok"}

if session.has_errors():
    for err in session.get_errors():
        print(f"{err.status}: {err.message}")

from mlsdk import Client, APIResponse

def ml_error_reporter(err: Exception):
    print(str(err))

client = Client(...)
session = client.create_session(on_error=ml_error_reporter)

from mlsdk import Client

client = Client(api_key="KEY", project_id="ID")

from mlsdk import Client

try:
    await client.user_identify(
        id="JJ@mail.com",
        traits={
            "name": "Jacob Jones",
            "email": "jj@mail.com",
            "country": "United States"
        }
    )
except Exception as (e):
    print(e.message)

from mlsdk import Client

try:
    await client.user_alias(
        id="jjacob",
        previous_id="JJ@mail.com",
    )
except Exception as (e):
    print(e.message)

session = client.create_session(
    session_id=str(uuid.uuid4()),
    conversation_id=str(uuid.uuid4()),
    id='jjacob'
)

await session.track_event(event="Start Chat", properties={"from": "shopping cart"})
await session.track_conversation_turn(
    user="I need help choosing the right lipstick for my skin color.",
    assistant="I can help you with that.  What color would you use to describe your skin tone?",
await session.end_session()
await session.flush()

conversation_id_1 = str(uuid.uuid4())
conversation_id_2 = str(uuid.uuid4())

session = client.create_session(
    session_id=str(uuid.uuid4()),
    id='jjacob'
)

# custom events do not have to be associated with a conversation.
await session.track_event(event="Start Chat", properties={"from": "shopping cart"})

await session.track_conversation_turn(
    conversation_id=conversation_id_1,
    user="I need help choosing the right lipstick for my skin color.",
    assistant="I can help you with that.  What color would you use to describe your skin tone?",

await session.track_conversation_turn(
    conversation_id=conversation_id_2,
    user="Do we sell lipstick?",
    assistant="Yes",
)

await session.end_conversation(conversation_id=conversation_id_2)
await session.end_conversation(conversation_id=conversation_id_1)

await session.flush()

await session.flush()

await session.flush_and_continue()

await session.end_session()

if session.has_errors():
    errors = session.get_errors()

class APIResponse(BaseModel):
    """Base class for API responses.

    Attributes:
        errored (bool): Indicates if the API response contains an error.
        status (str): The status of the API response.
        message (str): A message associated with the API response.
    """

    errored: bool
    status: int
    message: str

await session.user_identify(
    id="JJ@mail.com",
    traits={
        "name": "Jacob Jones",
        "email": "jj@mail.com",
        "country": "United States"
    }
)

await session.user_alias(
    id="jjacob",
    previous_id="JJ@mail.com",
)

await session.track_event(event="My Custom Event")
await session.track_event(
    event="Another Event",
    properties={
        "email": "test@test.com", # str
        "age": 30,                # int
        "is_subscribed": True,    # bool
        "height": 1.75,           # float
    }
)

await session.end_conversation()

await session.track_conversation_turn(
    user="I am feeling hungry so I would like to find a place to eat.",
    assistant="Do you have a specific which you want the eating place to be located at?"
)

class TokenBasedCost(BaseModel):
    """Common models have costs that are provided by a service on the web.

    If you are using one of these models, you can provide the model name and the
    number of tokens in the prompt and completion, and the cost will be calculated for you.
    """

    model: str = Field(..., min_length=1, max_length=100)
    prompt_tokens: int
    completion_tokens: int


class Cost(BaseModel):
    """If you know the cost of a conversation turn, you can provide it directly.

    This will be accumulated in the conversation analysis.
    """

    cost: float

await session.track_conversation_usage(
    cost: TokenBasedCost(model="gpt-4o", prompt_tokens=134, completion_tokens=237)
)

await session.track_function_call(
    name="my_function_name",
    args='{"input1": 5, "input2": 6}',
    result="17",
    runtime=4093
)

from mlsdk import HTTPClient

client = HTTPClient(
    api_key="YOUR_WORKSPACE_API_KEY",
    project_id="YOUR_PROJECT_ID",
)

response = await send_request(
    url="/bc/v1/events/queue",
    method="POST",
    data={
        # your data
    }
)

{
    "errored": True, # or False
    "status": 500,   # http status code
    "message": "..." # Error message
}

from mlsdk import Client, MLEvent

async def main():
    client = Client(
        api_key="YOUR_WORKSPACE_API_KEY",
        project_id="YOUR_PROJECT_ID",
    )

    async def on_event(event: MLEvent) -> None:
        print(f"Received event: {event}")

    async def on_error(error: Exception) -> None:
        print(f"Error: {error}")

    session_context = client.create_session(
        device_id="test_device_id",
        on_event=on_event,
        on_error=on_error,
    )

    async with session_context as session:
        await session.track_conversation_turn(
            user="I would like book an airline flight to New York.",
            assistant="No problem!  When would you like to arrive?",
        )
    # leaving the context will automatically flush any pending data in the queue and wait until
    # everything has been sent.  Because you registered callbacks for websockets, the websocket connection
    # will wait until a "Session Ended" event arrives, and then close down the websocket connection.

asyncio.run(main())

poetry run python -m ipykernel install --user --name=mindlytics --display-name "Mindlytics Python SDK"

OPENAI_API_KEY="yours"
MLSDK_API_KEY="yours"
MLSDK_PROJECT_ID="yours"

MLSDK_SERVER_BASE="http://localhost:3000"

eval `cat .env.examples` poetry run jupyter lab examples