akson package

akson package contains the Assistant interface that needs to be implemented by assistants.

Assistant

Bases: ABC

Assistants are used to generate responses to chats.

Source code in akson.py

class Assistant(ABC):
    """Assistants are used to generate responses to chats."""

    def __init__(self):
        self.name: str = self.__class__.__name__
        """Name of the assistant. Visible in the UI."""
        self.description: Optional[str] = self.__class__.__doc__
        """Description of the assistant, its purpose and capabilities."""

    def __repr__(self):
        return f"Assistant<{self.name}>"

    @abstractmethod
    async def run(self, chat: Chat) -> None:
        """
        Run the assistant on the given chat.
        This method will be called by the web server.
        """
        ...

description = self.__class__.__doc__ instance-attribute

Description of the assistant, its purpose and capabilities.

name = self.__class__.__name__ instance-attribute

Name of the assistant. Visible in the UI.

run(chat) abstractmethod async

Run the assistant on the given chat. This method will be called by the web server.

Source code in akson.py

@abstractmethod
async def run(self, chat: Chat) -> None:
    """
    Run the assistant on the given chat.
    This method will be called by the web server.
    """
    ...

Chat

Chat holds state and handles sending and receiving messages. It is passed to the assistant's run method. This serves as the main interface between the assistant and the web application.

Source code in akson.py

class Chat:
    """
    Chat holds state and handles sending and receiving messages.
    It is passed to the assistant's run method.
    This serves as the main interface between the assistant and the web application.
    """

    def __init__(
        self,
        *,
        state: Optional[ChatState] = None,
        publisher: Optional[Callable[[dict], Coroutine]] = None,
    ):
        if not state:
            state = ChatState()

        # Holds the chat's persistent state loaded from disk.
        # Mainly includes the history of chat messages, which is a list of Message.
        self.state = state
        """Holds the chat's persistent state loaded from disk."""

        # Contains new messages generated during the assistant run.
        self.new_messages: list[Message] = []

        # Publishes messages to clients.
        self.publisher = publisher

    async def reply(self, role: Literal["assistant", "tool"], name: str) -> Reply:
        # category: Optional[Literal["info", "success", "warning", "error"]] = None,
        return await Reply.create(chat=self, role=role, name=name)

    async def _queue_message(self, message: dict):
        if self.publisher:
            await self.publisher(message)

state = state instance-attribute

Holds the chat's persistent state loaded from disk.

ChatState pydantic-model

Bases: BaseModel

Chat that can be saved and loaded from a file.

Show JSON schema:

{
  "$defs": {
    "Message": {
      "properties": {
        "id": {
          "title": "Id",
          "type": "string"
        },
        "role": {
          "enum": [
            "user",
            "assistant",
            "tool"
          ],
          "title": "Role",
          "type": "string"
        },
        "name": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Name"
        },
        "content": {
          "title": "Content",
          "type": "string"
        },
        "tool_call": {
          "anyOf": [
            {
              "$ref": "#/$defs/ToolCall"
            },
            {
              "type": "null"
            }
          ],
          "default": null
        },
        "tool_call_id": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Tool Call Id"
        }
      },
      "required": [
        "role",
        "content"
      ],
      "title": "Message",
      "type": "object"
    },
    "ToolCall": {
      "properties": {
        "id": {
          "title": "Id",
          "type": "string"
        },
        "name": {
          "title": "Name",
          "type": "string"
        },
        "arguments": {
          "title": "Arguments",
          "type": "string"
        }
      },
      "required": [
        "id",
        "name",
        "arguments"
      ],
      "title": "ToolCall",
      "type": "object"
    }
  },
  "description": "Chat that can be saved and loaded from a file.",
  "properties": {
    "id": {
      "title": "Id",
      "type": "string"
    },
    "messages": {
      "default": [],
      "items": {
        "$ref": "#/$defs/Message"
      },
      "title": "Messages",
      "type": "array"
    },
    "assistant": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Assistant"
    },
    "title": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Title"
    }
  },
  "title": "ChatState",
  "type": "object"
}

Fields:

• id (str)
• messages (list[Message])
• assistant (Optional[str])
• title (Optional[str])

Source code in akson.py

class ChatState(BaseModel):
    """Chat that can be saved and loaded from a file."""

    id: str = Field(default_factory=generate_chat_id)
    messages: list[Message] = []
    assistant: Optional[str] = None
    title: Optional[str] = None

    @classmethod
    def create_new(cls, id: str, assistant: str):
        return cls(id=id, assistant=assistant)

    # TODO make this instance method
    @classmethod
    def load_from_disk(cls, chat_id: str):
        with open(cls.file_path(chat_id), "r") as f:
            content = f.read()
            return cls.model_validate_json(content)

    def save_to_disk(self):
        os.makedirs("chats", exist_ok=True)
        with open(self.file_path(self.id), "w") as f:
            f.write(self.model_dump_json(indent=2))

    @staticmethod
    def file_path(id: str):
        return os.path.join("chats", f"{id}.json")