Session

Session

BaseMessage

director.core.session.BaseMessage pydantic-model

Bases: BaseModel

Base message class for the input/output message. All the input/output messages will be inherited from this class.

Config:

• arbitrary_types_allowed: True
• use_enum_values: True
• validate_default: True

Fields:

• session_id (str)
• conv_id (str)
• msg_type (MsgType)
• actions (List[str])
• agents (List[str])
• content (List[Union[dict, TextContent, ImageContent, VideoContent, VideosContent, SearchResultsContent]])
• status (MsgStatus)
• msg_id (str)

InputMessage

director.core.session.InputMessage pydantic-model

Bases: BaseMessage

Input message from the user. This class is used to create the input message from the user.

publish

publish()

Store the message in the database. for conversation history.

Source code in backend/director/core/session.py

def publish(self):
    """Store the message in the database. for conversation history."""
    self.db.add_or_update_msg_to_conv(**self.model_dump(exclude={"db"}))

Output Message

director.core.session.OutputMessage pydantic-model

Bases: BaseMessage

Output message from the director. This class is used to create the output message from the director.

Fields:

• db (BaseDB)
• msg_type (MsgType)
• status (MsgStatus)

update_status

update_status(status)

Update the status of the message and publish the message to the socket. for loading state.

Source code in backend/director/core/session.py

def update_status(self, status: MsgStatus):
    """Update the status of the message and publish the message to the socket. for loading state."""
    self.status = status
    self._publish()

push_update

push_update()

Publish the message to the socket.

Source code in backend/director/core/session.py

def push_update(self):
    """Publish the message to the socket."""
    try:
        self._publish()
    except Exception as e:
        print(f"Error in emitting message: {str(e)}")

publish

publish()

Store the message in the database. for conversation history and publish the message to the socket.

Source code in backend/director/core/session.py

def publish(self):
    """Store the message in the database. for conversation history and publish the message to the socket."""
    self._publish()

Context Message

director.core.session.ContextMessage pydantic-model

Bases: BaseModel

Context message class. This class is used to create the context message for the reasoning context.

Config:

• arbitrary_types_allowed: True
• validate_default: True
• use_enum_values: True

Fields:

• content (Optional[Union[List[dict], str]])
• tool_calls (Optional[List[dict]])
• tool_call_id (Optional[str])
• role (RoleTypes)

to_llm_msg

to_llm_msg()

Convert the context message to the llm message.

Source code in backend/director/core/session.py

def to_llm_msg(self):
    """Convert the context message to the llm message."""
    msg = {
        "role": self.role,
        "content": self.content,
    }
    if self.role == RoleTypes.system:
        return msg

    if self.role == RoleTypes.user:
        return format_user_message(msg)

    if self.role == RoleTypes.assistant:
        if self.tool_calls:
            msg["tool_calls"] = self.tool_calls
        if not self.content:
            msg["content"] = []
        return msg

    if self.role == RoleTypes.tool:
        msg["tool_call_id"] = self.tool_call_id
        return msg

from_json classmethod

from_json(json_data)

Create the context message from the json data.

Source code in backend/director/core/session.py

@classmethod
def from_json(cls, json_data):
    """Create the context message from the json data."""
    return cls(**json_data)

Session

director.core.session.Session

Session(
    db,
    session_id="",
    conv_id="",
    collection_id=None,
    video_id=None,
    **kwargs
)

A class to manage and interact with a session in the database. The session is used to store the conversation and reasoning context messages.

Source code in backend/director/core/session.py

def __init__(
    self,
    db: BaseDB,
    session_id: str = "",
    conv_id: str = "",
    collection_id: str = None,
    video_id: str = None,
    **kwargs,
):
    self.db = db
    self.session_id = session_id
    self.conv_id = conv_id
    self.conversations = []
    self.video_id = video_id
    self.collection_id = collection_id
    self.reasoning_context = []
    self.state = {}
    self.output_message = OutputMessage(
        db=self.db, session_id=self.session_id, conv_id=self.conv_id
    )

    self.get_context_messages()

save_context_messages

save_context_messages()

Save the reasoning context messages to the database.

Source code in backend/director/core/session.py

def save_context_messages(self):
    """Save the reasoning context messages to the database."""
    context = {
        "reasoning": [message.to_llm_msg() for message in self.reasoning_context],
    }
    self.db.add_or_update_context_msg(self.session_id, context)

get_context_messages

get_context_messages()

Get the reasoning context messages from the database.

Source code in backend/director/core/session.py

def get_context_messages(self):
    """Get the reasoning context messages from the database."""
    if not self.reasoning_context:
        context = self.db.get_context_messages(self.session_id)
        self.reasoning_context = [
            ContextMessage.from_json(message)
            for message in context.get("reasoning", [])
        ]

    return self.reasoning_context

create

create()

Create a new session in the database.

Source code in backend/director/core/session.py

def create(self):
    """Create a new session in the database."""
    self.db.create_session(**self.__dict__)

new_message

new_message(msg_type=MsgType.output, **kwargs)

Returns a new input/output message object.

Parameters:

Name	Type	Description	Default
msg_type	MsgType	The type of the message, input or output.	output
kwargs	dict	The message attributes.	{}

Returns:

Type	Description
Union[InputMessage, OutputMessage]	The input/output message object.

Source code in backend/director/core/session.py

def new_message(
    self, msg_type: MsgType = MsgType.output, **kwargs
) -> Union[InputMessage, OutputMessage]:
    """Returns a new input/output message object.

    :param MsgType msg_type: The type of the message, input or output.
    :param dict kwargs: The message attributes.
    :return: The input/output message object.
    """
    if msg_type == MsgType.input:
        return InputMessage(
            db=self.db,
            session_id=self.session_id,
            conv_id=self.conv_id,
            **kwargs,
        )
    return OutputMessage(
        db=self.db,
        session_id=self.session_id,
        conv_id=self.conv_id,
        **kwargs,
    )

get

get()

Get the session from the database.

Source code in backend/director/core/session.py

def get(self):
    """Get the session from the database."""
    session = self.db.get_session(self.session_id)
    conversation = self.db.get_conversations(self.session_id)
    session["conversation"] = conversation
    return session

get_all

get_all()

Get all the sessions from the database.

Source code in backend/director/core/session.py

def get_all(self):
    """Get all the sessions from the database."""
    return self.db.get_sessions()

delete

delete()

Delete the session from the database.

Source code in backend/director/core/session.py

def delete(self):
    """Delete the session from the database."""
    return self.db.delete_session(self.session_id)

emit_event

emit_event(event, namespace='/chat')

Emits a structured WebSocket event to notify all clients about updates.

Source code in backend/director/core/session.py

def emit_event(self, event: BaseEvent, namespace="/chat"):
    """Emits a structured WebSocket event to notify all clients about updates."""

    event_payload = event.model_dump()

    try:
        emit("event", event_payload, namespace=namespace)
    except Exception:
        pass