Skip to content

Thread schema

ThreadBase

Bases: BaseModel

Base schema for a Thread.

Source code in light_agents/schemas/thread_schema.py 
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
class ThreadBase(BaseModel):
    """Base schema for a Thread."""

    model_config = model_config
    type: ThreadType
    messages: MutableSequence[MessageBase] = []
    external_thread_fields: Dict[str, Any] = {}

    def add_message(self, message: MessageBase) -> None:
        """Add a message to the thread."""
        if not isinstance(message, MessageBase):
            raise ValueError("The message must be an instance of MessageBase.")

        self.messages.append(message)

    def add_messages_list(
        self, messages: Sequence[MessageBase]
    ) -> None:
        """Add a list of messages to the thread."""
        if not all(isinstance(message, MessageBase) for message in messages):
            raise ValueError("All messages must be instances of MessageBase.")

        for message in messages:
            self.add_message(message)

    def process_thread(self, thread_agent: ThreadAgent) -> None:
        """Process the thread."""
        if not isinstance(thread_agent, ThreadAgent):
            raise ValueError(
                "The thread agent must be an instance of ThreadAgent."
            )

        ## TODO:
        ## 1. Deal gracefully with exceptions
        ## 2. Update external thread fields based on the agent's output
        messages = thread_agent.agent_run(
            self.messages, **self.external_thread_fields
        )
        self.add_messages_list(messages)

add_message(message)

Add a message to the thread.

Source code in light_agents/schemas/thread_schema.py 
25
26
27
28
29
30
def add_message(self, message: MessageBase) -> None:
    """Add a message to the thread."""
    if not isinstance(message, MessageBase):
        raise ValueError("The message must be an instance of MessageBase.")

    self.messages.append(message)

add_messages_list(messages)

Add a list of messages to the thread.

Source code in light_agents/schemas/thread_schema.py 
32
33
34
35
36
37
38
39
40
def add_messages_list(
    self, messages: Sequence[MessageBase]
) -> None:
    """Add a list of messages to the thread."""
    if not all(isinstance(message, MessageBase) for message in messages):
        raise ValueError("All messages must be instances of MessageBase.")

    for message in messages:
        self.add_message(message)

process_thread(thread_agent)

Process the thread.

Source code in light_agents/schemas/thread_schema.py 
42
43
44
45
46
47
48
49
50
51
52
53
54
55
def process_thread(self, thread_agent: ThreadAgent) -> None:
    """Process the thread."""
    if not isinstance(thread_agent, ThreadAgent):
        raise ValueError(
            "The thread agent must be an instance of ThreadAgent."
        )

    ## TODO:
    ## 1. Deal gracefully with exceptions
    ## 2. Update external thread fields based on the agent's output
    messages = thread_agent.agent_run(
        self.messages, **self.external_thread_fields
    )
    self.add_messages_list(messages)

ThreadType

Bases: str, Enum

Possible types for a single Thread.

Source code in light_agents/schemas/thread_schema.py 
11
12
13
14
class ThreadType(str, Enum):
    """Possible types for a single Thread."""

    BASIC = "basic"