Skip to content

common

__all__ = ['ActionCallDeltaMessageContent', 'ActionCallMessageContent', 'ActionResultMessageContent', 'AudioDeltaMessageContent', 'AudioMessageContent', 'BaseAction', 'BaseDeltaMessageContent', 'BaseMessage', 'BaseMessageContent', 'DeltaMessage', 'GenericMessageContent', 'ImageMessageContent', 'Message', 'Observable', 'PromptStack', 'Reference', 'TextDeltaMessageContent', 'TextMessageContent', 'ToolAction', 'observable'] module-attribute

ActionCallDeltaMessageContent

Bases: BaseDeltaMessageContent

Source code in griptape/common/prompt_stack/contents/action_call_delta_message_content.py 
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
@define
class ActionCallDeltaMessageContent(BaseDeltaMessageContent):
    tag: Optional[str] = field(default=None, metadata={"serializable": True})
    name: Optional[str] = field(default=None, metadata={"serializable": True})
    path: Optional[str] = field(default=None, metadata={"serializable": True})
    partial_input: Optional[str] = field(default=None, metadata={"serializable": True})

    def __str__(self) -> str:
        parts = []

        if self.name:
            parts.append(self.name)
            if self.path:
                parts.append(f".{self.path}")
                if self.tag:
                    parts.append(f" ({self.tag})")

        if self.partial_input:
            if parts:
                parts.append(f" {self.partial_input}")
            else:
                parts.append(self.partial_input)

        return "".join(parts)

name = field(default=None, metadata={'serializable': True}) class-attribute instance-attribute

partial_input = field(default=None, metadata={'serializable': True}) class-attribute instance-attribute

path = field(default=None, metadata={'serializable': True}) class-attribute instance-attribute

tag = field(default=None, metadata={'serializable': True}) class-attribute instance-attribute

__str__()

Source code in griptape/common/prompt_stack/contents/action_call_delta_message_content.py 
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
def __str__(self) -> str:
    parts = []

    if self.name:
        parts.append(self.name)
        if self.path:
            parts.append(f".{self.path}")
            if self.tag:
                parts.append(f" ({self.tag})")

    if self.partial_input:
        if parts:
            parts.append(f" {self.partial_input}")
        else:
            parts.append(self.partial_input)

    return "".join(parts)

ActionCallMessageContent

Bases: BaseMessageContent

Source code in griptape/common/prompt_stack/contents/action_call_message_content.py 
15
16
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
@define
class ActionCallMessageContent(BaseMessageContent):
    artifact: ActionArtifact = field(metadata={"serializable": True})

    @classmethod
    def from_deltas(cls, deltas: Sequence[BaseDeltaMessageContent]) -> ActionCallMessageContent:
        action_call_deltas = [delta for delta in deltas if isinstance(delta, ActionCallDeltaMessageContent)]

        tag = None
        name = None
        path = None
        json_input = ""

        for delta in action_call_deltas:
            if delta.tag is not None:
                tag = delta.tag
            if delta.name is not None:
                name = delta.name
            if delta.path is not None:
                path = delta.path
            if delta.partial_input is not None:
                json_input += delta.partial_input

        if tag is not None and name is not None and path is not None:
            try:
                parsed_input = json.loads(json_input or "{}")
            except json.JSONDecodeError as exc:
                raise ValueError("Invalid JSON input for ToolAction") from exc
            action = ToolAction(tag=tag, name=name, path=path, input=parsed_input)
        else:
            raise ValueError("Missing required fields for ToolAction")

        artifact = ActionArtifact(value=action)

        return cls(artifact=artifact)

artifact = field(metadata={'serializable': True}) class-attribute instance-attribute

from_deltas(deltas) classmethod

Source code in griptape/common/prompt_stack/contents/action_call_message_content.py 
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
@classmethod
def from_deltas(cls, deltas: Sequence[BaseDeltaMessageContent]) -> ActionCallMessageContent:
    action_call_deltas = [delta for delta in deltas if isinstance(delta, ActionCallDeltaMessageContent)]

    tag = None
    name = None
    path = None
    json_input = ""

    for delta in action_call_deltas:
        if delta.tag is not None:
            tag = delta.tag
        if delta.name is not None:
            name = delta.name
        if delta.path is not None:
            path = delta.path
        if delta.partial_input is not None:
            json_input += delta.partial_input

    if tag is not None and name is not None and path is not None:
        try:
            parsed_input = json.loads(json_input or "{}")
        except json.JSONDecodeError as exc:
            raise ValueError("Invalid JSON input for ToolAction") from exc
        action = ToolAction(tag=tag, name=name, path=path, input=parsed_input)
    else:
        raise ValueError("Missing required fields for ToolAction")

    artifact = ActionArtifact(value=action)

    return cls(artifact=artifact)

ActionResultMessageContent

Bases: BaseMessageContent

Source code in griptape/common/prompt_stack/contents/action_result_message_content.py 
15
16
17
18
19
20
21
22
@define
class ActionResultMessageContent(BaseMessageContent):
    artifact: BaseArtifact = field(metadata={"serializable": True})
    action: ToolAction = field(metadata={"serializable": True})

    @classmethod
    def from_deltas(cls, deltas: Sequence[BaseDeltaMessageContent]) -> ActionResultMessageContent:
        raise NotImplementedError

action = field(metadata={'serializable': True}) class-attribute instance-attribute

artifact = field(metadata={'serializable': True}) class-attribute instance-attribute

from_deltas(deltas) classmethod

Source code in griptape/common/prompt_stack/contents/action_result_message_content.py 
20
21
22
@classmethod
def from_deltas(cls, deltas: Sequence[BaseDeltaMessageContent]) -> ActionResultMessageContent:
    raise NotImplementedError

AudioDeltaMessageContent

Bases: BaseDeltaMessageContent

A delta message content for audio data.

Attributes:

Name Type Description
id Optional[str]

The ID of the audio data.
data Optional[str]

Base64 encoded audio data.
transcript Optional[str]

The transcript of the audio data.
expires_at Optional[int]

The Unix timestamp (in seconds) for when this audio data will no longer be accessible.
Source code in griptape/common/prompt_stack/contents/audio_delta_message_content.py 
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
@define(kw_only=True)
class AudioDeltaMessageContent(BaseDeltaMessageContent):
    """A delta message content for audio data.

    Attributes:
        id: The ID of the audio data.
        data: Base64 encoded audio data.
        transcript: The transcript of the audio data.
        expires_at: The Unix timestamp (in seconds) for when this audio data will no longer be accessible.
    """

    id: Optional[str] = field(default=None, metadata={"serializable": True})
    data: Optional[str] = field(default=None, metadata={"serializable": True})
    transcript: Optional[str] = field(default=None, metadata={"serializable": True})
    expires_at: Optional[int] = field(default=None, metadata={"serializable": True})

data = field(default=None, metadata={'serializable': True}) class-attribute instance-attribute

expires_at = field(default=None, metadata={'serializable': True}) class-attribute instance-attribute

id = field(default=None, metadata={'serializable': True}) class-attribute instance-attribute

transcript = field(default=None, metadata={'serializable': True}) class-attribute instance-attribute

AudioMessageContent

Bases: BaseMessageContent

Source code in griptape/common/prompt_stack/contents/audio_message_content.py 
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
@define
class AudioMessageContent(BaseMessageContent):
    artifact: AudioArtifact = field(metadata={"serializable": True})

    @classmethod
    def from_deltas(cls, deltas: Sequence[BaseDeltaMessageContent]) -> AudioMessageContent:
        audio_deltas = [delta for delta in deltas if isinstance(delta, AudioDeltaMessageContent)]
        audio_data = [delta.data for delta in audio_deltas if delta.data is not None]
        transcript_data = [delta.transcript for delta in audio_deltas if delta.transcript is not None]
        expires_at = next(delta.expires_at for delta in audio_deltas if delta.expires_at is not None)
        audio_id = next(delta.id for delta in audio_deltas if delta.id is not None)

        audio_transcript = "".join(data for data in transcript_data)

        artifact = AudioArtifact(
            value=b"".join(base64.b64decode(data) for data in audio_data),
            format="wav",
            meta={
                "audio_id": audio_id,
                "expires_at": expires_at,
                "transcript": audio_transcript,
            },
        )

        return cls(artifact=artifact)

artifact = field(metadata={'serializable': True}) class-attribute instance-attribute

from_deltas(deltas) classmethod

Source code in griptape/common/prompt_stack/contents/audio_message_content.py 
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
@classmethod
def from_deltas(cls, deltas: Sequence[BaseDeltaMessageContent]) -> AudioMessageContent:
    audio_deltas = [delta for delta in deltas if isinstance(delta, AudioDeltaMessageContent)]
    audio_data = [delta.data for delta in audio_deltas if delta.data is not None]
    transcript_data = [delta.transcript for delta in audio_deltas if delta.transcript is not None]
    expires_at = next(delta.expires_at for delta in audio_deltas if delta.expires_at is not None)
    audio_id = next(delta.id for delta in audio_deltas if delta.id is not None)

    audio_transcript = "".join(data for data in transcript_data)

    artifact = AudioArtifact(
        value=b"".join(base64.b64decode(data) for data in audio_data),
        format="wav",
        meta={
            "audio_id": audio_id,
            "expires_at": expires_at,
            "transcript": audio_transcript,
        },
    )

    return cls(artifact=artifact)

BaseAction

Bases: SerializableMixin, ABC

Source code in griptape/common/actions/base_action.py 
6
class BaseAction(SerializableMixin, ABC): ...

BaseDeltaMessageContent

Bases: ABC, SerializableMixin

Source code in griptape/common/prompt_stack/contents/base_delta_message_content.py 
10
11
12
@define
class BaseDeltaMessageContent(ABC, SerializableMixin):
    index: int = field(kw_only=True, default=0, metadata={"serializable": True})

index = field(kw_only=True, default=0, metadata={'serializable': True}) class-attribute instance-attribute

BaseMessage

Bases: ABC, SerializableMixin

Source code in griptape/common/prompt_stack/messages/base_message.py 
14
15
16
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
@define
class BaseMessage(ABC, SerializableMixin):
    @define
    class Usage(SerializableMixin):
        input_tokens: Optional[float] = field(kw_only=True, default=None, metadata={"serializable": True})
        output_tokens: Optional[float] = field(kw_only=True, default=None, metadata={"serializable": True})

        @property
        def total_tokens(self) -> float:
            return (self.input_tokens or 0) + (self.output_tokens or 0)

        def __add__(self, other: BaseMessage.Usage) -> BaseMessage.Usage:
            return BaseMessage.Usage(
                input_tokens=(self.input_tokens or 0) + (other.input_tokens or 0),
                output_tokens=(self.output_tokens or 0) + (other.output_tokens or 0),
            )

    USER_ROLE = "user"
    ASSISTANT_ROLE = "assistant"
    SYSTEM_ROLE = "system"

    content: list[Union[BaseMessageContent, BaseDeltaMessageContent]] = field(metadata={"serializable": True})
    role: str = field(kw_only=True, metadata={"serializable": True})
    usage: Usage = field(kw_only=True, default=Factory(lambda: BaseMessage.Usage()), metadata={"serializable": True})

    def is_system(self) -> bool:
        return self.role == self.SYSTEM_ROLE

    def is_user(self) -> bool:
        return self.role == self.USER_ROLE

    def is_assistant(self) -> bool:
        return self.role == self.ASSISTANT_ROLE

ASSISTANT_ROLE = 'assistant' class-attribute instance-attribute

SYSTEM_ROLE = 'system' class-attribute instance-attribute

USER_ROLE = 'user' class-attribute instance-attribute

content = field(metadata={'serializable': True}) class-attribute instance-attribute

role = field(kw_only=True, metadata={'serializable': True}) class-attribute instance-attribute

usage = field(kw_only=True, default=Factory(lambda: BaseMessage.Usage()), metadata={'serializable': True}) class-attribute instance-attribute

Usage

Bases: SerializableMixin

Source code in griptape/common/prompt_stack/messages/base_message.py 
16
17
18
19
20
21
22
23
24
25
26
27
28
29
@define
class Usage(SerializableMixin):
    input_tokens: Optional[float] = field(kw_only=True, default=None, metadata={"serializable": True})
    output_tokens: Optional[float] = field(kw_only=True, default=None, metadata={"serializable": True})

    @property
    def total_tokens(self) -> float:
        return (self.input_tokens or 0) + (self.output_tokens or 0)

    def __add__(self, other: BaseMessage.Usage) -> BaseMessage.Usage:
        return BaseMessage.Usage(
            input_tokens=(self.input_tokens or 0) + (other.input_tokens or 0),
            output_tokens=(self.output_tokens or 0) + (other.output_tokens or 0),
        )
input_tokens = field(kw_only=True, default=None, metadata={'serializable': True}) class-attribute instance-attribute
output_tokens = field(kw_only=True, default=None, metadata={'serializable': True}) class-attribute instance-attribute
total_tokens property
__add__(other)
Source code in griptape/common/prompt_stack/messages/base_message.py 
25
26
27
28
29
def __add__(self, other: BaseMessage.Usage) -> BaseMessage.Usage:
    return BaseMessage.Usage(
        input_tokens=(self.input_tokens or 0) + (other.input_tokens or 0),
        output_tokens=(self.output_tokens or 0) + (other.output_tokens or 0),
    )

is_assistant()

Source code in griptape/common/prompt_stack/messages/base_message.py 
45
46
def is_assistant(self) -> bool:
    return self.role == self.ASSISTANT_ROLE

is_system()

Source code in griptape/common/prompt_stack/messages/base_message.py 
39
40
def is_system(self) -> bool:
    return self.role == self.SYSTEM_ROLE

is_user()

Source code in griptape/common/prompt_stack/messages/base_message.py 
42
43
def is_user(self) -> bool:
    return self.role == self.USER_ROLE

BaseMessageContent

Bases: ABC, SerializableMixin

Source code in griptape/common/prompt_stack/contents/base_message_content.py 
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
@define
class BaseMessageContent(ABC, SerializableMixin):
    artifact: BaseArtifact = field(metadata={"serializable": True})

    def __str__(self) -> str:
        return self.artifact.to_text()

    def __bool__(self) -> bool:
        return bool(self.artifact)

    def __len__(self) -> int:
        return len(self.artifact)

    def to_text(self) -> str:
        return str(self.artifact)

    @classmethod
    @abstractmethod
    def from_deltas(cls, deltas: Sequence[BaseDeltaMessageContent]) -> BaseMessageContent: ...

artifact = field(metadata={'serializable': True}) class-attribute instance-attribute

__bool__()

Source code in griptape/common/prompt_stack/contents/base_message_content.py 
25
26
def __bool__(self) -> bool:
    return bool(self.artifact)

__len__()

Source code in griptape/common/prompt_stack/contents/base_message_content.py 
28
29
def __len__(self) -> int:
    return len(self.artifact)

__str__()

Source code in griptape/common/prompt_stack/contents/base_message_content.py 
22
23
def __str__(self) -> str:
    return self.artifact.to_text()

from_deltas(deltas) abstractmethod classmethod

Source code in griptape/common/prompt_stack/contents/base_message_content.py 
34
35
36
@classmethod
@abstractmethod
def from_deltas(cls, deltas: Sequence[BaseDeltaMessageContent]) -> BaseMessageContent: ...

to_text()

Source code in griptape/common/prompt_stack/contents/base_message_content.py 
31
32
def to_text(self) -> str:
    return str(self.artifact)

DeltaMessage

Bases: BaseMessage

Source code in griptape/common/prompt_stack/messages/delta_message.py 
13
14
15
16
@define
class DeltaMessage(BaseMessage):
    role: Optional[str] = field(kw_only=True, default=None, metadata={"serializable": True})
    content: Optional[BaseDeltaMessageContent] = field(kw_only=True, default=None, metadata={"serializable": True})

content = field(kw_only=True, default=None, metadata={'serializable': True}) class-attribute instance-attribute

role = field(kw_only=True, default=None, metadata={'serializable': True}) class-attribute instance-attribute

GenericMessageContent

Bases: BaseMessageContent

Source code in griptape/common/prompt_stack/contents/generic_message_content.py 
15
16
17
18
19
20
21
@define
class GenericMessageContent(BaseMessageContent):
    artifact: GenericArtifact = field(metadata={"serializable": True})

    @classmethod
    def from_deltas(cls, deltas: Sequence[BaseDeltaMessageContent]) -> GenericMessageContent:
        raise NotImplementedError

artifact = field(metadata={'serializable': True}) class-attribute instance-attribute

from_deltas(deltas) classmethod

Source code in griptape/common/prompt_stack/contents/generic_message_content.py 
19
20
21
@classmethod
def from_deltas(cls, deltas: Sequence[BaseDeltaMessageContent]) -> GenericMessageContent:
    raise NotImplementedError

ImageMessageContent

Bases: BaseMessageContent

Source code in griptape/common/prompt_stack/contents/image_message_content.py 
15
16
17
18
19
20
21
@define
class ImageMessageContent(BaseMessageContent):
    artifact: Union[ImageArtifact, ImageUrlArtifact] = field(metadata={"serializable": True})

    @classmethod
    def from_deltas(cls, deltas: Sequence[BaseDeltaMessageContent]) -> ImageMessageContent:
        raise NotImplementedError

artifact = field(metadata={'serializable': True}) class-attribute instance-attribute

from_deltas(deltas) classmethod

Source code in griptape/common/prompt_stack/contents/image_message_content.py 
19
20
21
@classmethod
def from_deltas(cls, deltas: Sequence[BaseDeltaMessageContent]) -> ImageMessageContent:
    raise NotImplementedError

Message

Bases: BaseMessage

Source code in griptape/common/prompt_stack/messages/message.py 
15
16
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
56
57
@define
class Message(BaseMessage):
    def __init__(self, content: str | list[BaseMessageContent], **kwargs: Any) -> None:
        if isinstance(content, str):
            content = [TextMessageContent(TextArtifact(value=content))]
        self.__attrs_init__(content, **kwargs)  # pyright: ignore[reportAttributeAccessIssue]

    content: list[BaseMessageContent] = field(metadata={"serializable": True})

    @property
    def value(self) -> Any:
        return self.to_artifact().value

    def __str__(self) -> str:
        return self.to_text()

    def has_all_content_type(self, content_type: type[T]) -> bool:
        return all(isinstance(content, content_type) for content in self.content)

    def has_any_content_type(self, content_type: type[T]) -> bool:
        return any(isinstance(content, content_type) for content in self.content)

    def get_content_type(self, content_type: type[T]) -> list[T]:
        return [content for content in self.content if isinstance(content, content_type)]

    def is_text(self) -> bool:
        return all(isinstance(content, TextMessageContent) for content in self.content)

    def to_text(self) -> str:
        return "".join(
            [content.artifact.to_text() for content in self.content if isinstance(content, TextMessageContent)],
        )

    def to_artifact(self, meta: Optional[dict] = None) -> BaseArtifact:
        if meta is None:
            meta = {}
        if len(self.content) == 1:
            artifact = self.content[0].artifact
        else:
            artifact = ListArtifact([content.artifact for content in self.content])

        artifact.meta.update(meta)
        return artifact

content = field(metadata={'serializable': True}) class-attribute instance-attribute

value property

__init__(content, **kwargs)

Source code in griptape/common/prompt_stack/messages/message.py 
17
18
19
20
def __init__(self, content: str | list[BaseMessageContent], **kwargs: Any) -> None:
    if isinstance(content, str):
        content = [TextMessageContent(TextArtifact(value=content))]
    self.__attrs_init__(content, **kwargs)  # pyright: ignore[reportAttributeAccessIssue]

__str__()

Source code in griptape/common/prompt_stack/messages/message.py 
28
29
def __str__(self) -> str:
    return self.to_text()

get_content_type(content_type)

Source code in griptape/common/prompt_stack/messages/message.py 
37
38
def get_content_type(self, content_type: type[T]) -> list[T]:
    return [content for content in self.content if isinstance(content, content_type)]

has_all_content_type(content_type)

Source code in griptape/common/prompt_stack/messages/message.py 
31
32
def has_all_content_type(self, content_type: type[T]) -> bool:
    return all(isinstance(content, content_type) for content in self.content)

has_any_content_type(content_type)

Source code in griptape/common/prompt_stack/messages/message.py 
34
35
def has_any_content_type(self, content_type: type[T]) -> bool:
    return any(isinstance(content, content_type) for content in self.content)

is_text()

Source code in griptape/common/prompt_stack/messages/message.py 
40
41
def is_text(self) -> bool:
    return all(isinstance(content, TextMessageContent) for content in self.content)

to_artifact(meta=None)

Source code in griptape/common/prompt_stack/messages/message.py 
48
49
50
51
52
53
54
55
56
57
def to_artifact(self, meta: Optional[dict] = None) -> BaseArtifact:
    if meta is None:
        meta = {}
    if len(self.content) == 1:
        artifact = self.content[0].artifact
    else:
        artifact = ListArtifact([content.artifact for content in self.content])

    artifact.meta.update(meta)
    return artifact

to_text()

Source code in griptape/common/prompt_stack/messages/message.py 
43
44
45
46
def to_text(self) -> str:
    return "".join(
        [content.artifact.to_text() for content in self.content if isinstance(content, TextMessageContent)],
    )

Observable

Source code in griptape/common/observable.py 
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
class Observable:
    @define
    class Call:
        func: Callable = field(kw_only=True)
        instance: Optional[Any] = field(default=None, kw_only=True)
        args: tuple[Any, ...] = field(default=Factory(tuple), kw_only=True)
        kwargs: dict[str, Any] = field(default=Factory(dict), kw_only=True)
        decorator_args: tuple[Any, ...] = field(default=Factory(tuple), kw_only=True)
        decorator_kwargs: dict[str, Any] = field(default=Factory(dict), kw_only=True)

        def __call__(self) -> Any:
            return self.func(*self.args, **self.kwargs)

        @property
        def tags(self) -> Optional[list[str]]:
            return self.decorator_kwargs.get("tags")

Call

Source code in griptape/common/observable.py 
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
@define
class Call:
    func: Callable = field(kw_only=True)
    instance: Optional[Any] = field(default=None, kw_only=True)
    args: tuple[Any, ...] = field(default=Factory(tuple), kw_only=True)
    kwargs: dict[str, Any] = field(default=Factory(dict), kw_only=True)
    decorator_args: tuple[Any, ...] = field(default=Factory(tuple), kw_only=True)
    decorator_kwargs: dict[str, Any] = field(default=Factory(dict), kw_only=True)

    def __call__(self) -> Any:
        return self.func(*self.args, **self.kwargs)

    @property
    def tags(self) -> Optional[list[str]]:
        return self.decorator_kwargs.get("tags")
args = field(default=Factory(tuple), kw_only=True) class-attribute instance-attribute
decorator_args = field(default=Factory(tuple), kw_only=True) class-attribute instance-attribute
decorator_kwargs = field(default=Factory(dict), kw_only=True) class-attribute instance-attribute
func = field(kw_only=True) class-attribute instance-attribute
instance = field(default=None, kw_only=True) class-attribute instance-attribute
kwargs = field(default=Factory(dict), kw_only=True) class-attribute instance-attribute
tags property
__call__()
Source code in griptape/common/observable.py 
18
19
def __call__(self) -> Any:
    return self.func(*self.args, **self.kwargs)

PromptStack

Bases: SerializableMixin

Source code in griptape/common/prompt_stack/prompt_stack.py 
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
@define
class PromptStack(SerializableMixin):
    messages: list[Message] = field(factory=list, kw_only=True, metadata={"serializable": True})
    tools: list[BaseTool] = field(factory=list, kw_only=True)
    output_schema: Optional[Union[Schema, type[BaseModel]]] = field(default=None, kw_only=True)

    @property
    def system_messages(self) -> list[Message]:
        return [message for message in self.messages if message.is_system()]

    @property
    def user_messages(self) -> list[Message]:
        return [message for message in self.messages if message.is_user()]

    @property
    def assistant_messages(self) -> list[Message]:
        return [message for message in self.messages if message.is_assistant()]

    def add_message(self, artifact: str | BaseArtifact, role: str) -> Message:
        content = self.__to_message_content(artifact)

        self.messages.append(Message(content=content, role=role))

        return self.messages[-1]

    def add_system_message(self, artifact: str | BaseArtifact) -> Message:
        return self.add_message(artifact, Message.SYSTEM_ROLE)

    def add_user_message(self, artifact: str | BaseArtifact) -> Message:
        return self.add_message(artifact, Message.USER_ROLE)

    def add_assistant_message(self, artifact: str | BaseArtifact) -> Message:
        return self.add_message(artifact, Message.ASSISTANT_ROLE)

    def to_output_json_schema(self, schema_id: str = "Output Format") -> dict:
        if self.output_schema is None:
            raise ValueError("Output schema is not set")

        if isinstance(self.output_schema, Schema):
            json_schema = self.output_schema.json_schema(schema_id)
        elif isinstance(self.output_schema, type) and issubclass(self.output_schema, BaseModel):
            json_schema = build_strict_schema(self.output_schema.model_json_schema(), schema_id)
        else:
            raise ValueError(f"Unsupported output schema type: {type(self.output_schema)}")

        return json_schema

    @classmethod
    def from_artifact(cls, artifact: BaseArtifact) -> PromptStack:
        prompt_stack = cls()
        prompt_stack.add_user_message(artifact)

        return prompt_stack

    def __to_message_content(self, artifact: str | BaseArtifact) -> list[BaseMessageContent]:
        if isinstance(artifact, str):
            return [TextMessageContent(TextArtifact(artifact))]
        if isinstance(artifact, TextArtifact):
            return [TextMessageContent(artifact)]
        if isinstance(artifact, ImageArtifact):
            return [ImageMessageContent(artifact)]
        if isinstance(artifact, ImageUrlArtifact):
            return [ImageMessageContent(artifact)]
        if isinstance(artifact, AudioArtifact):
            return [AudioMessageContent(artifact)]
        if isinstance(artifact, GenericArtifact):
            return [GenericMessageContent(artifact)]
        if isinstance(artifact, ActionArtifact):
            action = artifact.value
            output = action.output
            if output is None:
                return [ActionCallMessageContent(artifact)]
            return [ActionResultMessageContent(output, action=action)]
        if isinstance(artifact, ListArtifact):
            processed_contents = [self.__to_message_content(artifact) for artifact in artifact.value]

            return [sub_content for processed_content in processed_contents for sub_content in processed_content]
        return [TextMessageContent(TextArtifact(artifact.to_text()))]

assistant_messages property

messages = field(factory=list, kw_only=True, metadata={'serializable': True}) class-attribute instance-attribute

output_schema = field(default=None, kw_only=True) class-attribute instance-attribute

system_messages property

tools = field(factory=list, kw_only=True) class-attribute instance-attribute

user_messages property

__to_message_content(artifact)

Source code in griptape/common/prompt_stack/prompt_stack.py 
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
def __to_message_content(self, artifact: str | BaseArtifact) -> list[BaseMessageContent]:
    if isinstance(artifact, str):
        return [TextMessageContent(TextArtifact(artifact))]
    if isinstance(artifact, TextArtifact):
        return [TextMessageContent(artifact)]
    if isinstance(artifact, ImageArtifact):
        return [ImageMessageContent(artifact)]
    if isinstance(artifact, ImageUrlArtifact):
        return [ImageMessageContent(artifact)]
    if isinstance(artifact, AudioArtifact):
        return [AudioMessageContent(artifact)]
    if isinstance(artifact, GenericArtifact):
        return [GenericMessageContent(artifact)]
    if isinstance(artifact, ActionArtifact):
        action = artifact.value
        output = action.output
        if output is None:
            return [ActionCallMessageContent(artifact)]
        return [ActionResultMessageContent(output, action=action)]
    if isinstance(artifact, ListArtifact):
        processed_contents = [self.__to_message_content(artifact) for artifact in artifact.value]

        return [sub_content for processed_content in processed_contents for sub_content in processed_content]
    return [TextMessageContent(TextArtifact(artifact.to_text()))]

add_assistant_message(artifact)

Source code in griptape/common/prompt_stack/prompt_stack.py 
67
68
def add_assistant_message(self, artifact: str | BaseArtifact) -> Message:
    return self.add_message(artifact, Message.ASSISTANT_ROLE)

add_message(artifact, role)

Source code in griptape/common/prompt_stack/prompt_stack.py 
54
55
56
57
58
59
def add_message(self, artifact: str | BaseArtifact, role: str) -> Message:
    content = self.__to_message_content(artifact)

    self.messages.append(Message(content=content, role=role))

    return self.messages[-1]

add_system_message(artifact)

Source code in griptape/common/prompt_stack/prompt_stack.py 
61
62
def add_system_message(self, artifact: str | BaseArtifact) -> Message:
    return self.add_message(artifact, Message.SYSTEM_ROLE)

add_user_message(artifact)

Source code in griptape/common/prompt_stack/prompt_stack.py 
64
65
def add_user_message(self, artifact: str | BaseArtifact) -> Message:
    return self.add_message(artifact, Message.USER_ROLE)

from_artifact(artifact) classmethod

Source code in griptape/common/prompt_stack/prompt_stack.py 
83
84
85
86
87
88
@classmethod
def from_artifact(cls, artifact: BaseArtifact) -> PromptStack:
    prompt_stack = cls()
    prompt_stack.add_user_message(artifact)

    return prompt_stack

to_output_json_schema(schema_id='Output Format')

Source code in griptape/common/prompt_stack/prompt_stack.py 
70
71
72
73
74
75
76
77
78
79
80
81
def to_output_json_schema(self, schema_id: str = "Output Format") -> dict:
    if self.output_schema is None:
        raise ValueError("Output schema is not set")

    if isinstance(self.output_schema, Schema):
        json_schema = self.output_schema.json_schema(schema_id)
    elif isinstance(self.output_schema, type) and issubclass(self.output_schema, BaseModel):
        json_schema = build_strict_schema(self.output_schema.model_json_schema(), schema_id)
    else:
        raise ValueError(f"Unsupported output schema type: {type(self.output_schema)}")

    return json_schema

Reference

Bases: SerializableMixin

Source code in griptape/common/reference.py 
11
12
13
14
15
16
17
18
@define(kw_only=True)
class Reference(SerializableMixin):
    id: str = field(default=Factory(lambda: uuid.uuid4().hex), metadata={"serializable": True})
    title: str = field(metadata={"serializable": True})
    authors: list[str] = field(factory=list, metadata={"serializable": True})
    source: Optional[str] = field(default=None, metadata={"serializable": True})
    year: Optional[str] = field(default=None, metadata={"serializable": True})
    url: Optional[str] = field(default=None, metadata={"serializable": True})

authors = field(factory=list, metadata={'serializable': True}) class-attribute instance-attribute

id = field(default=Factory(lambda: uuid.uuid4().hex), metadata={'serializable': True}) class-attribute instance-attribute

source = field(default=None, metadata={'serializable': True}) class-attribute instance-attribute

title = field(metadata={'serializable': True}) class-attribute instance-attribute

url = field(default=None, metadata={'serializable': True}) class-attribute instance-attribute

year = field(default=None, metadata={'serializable': True}) class-attribute instance-attribute

TextDeltaMessageContent

Bases: BaseDeltaMessageContent

Source code in griptape/common/prompt_stack/contents/text_delta_message_content.py 
 8
 9
10
@define
class TextDeltaMessageContent(BaseDeltaMessageContent):
    text: str = field(metadata={"serializable": True})

text = field(metadata={'serializable': True}) class-attribute instance-attribute

TextMessageContent

Bases: BaseMessageContent

Source code in griptape/common/prompt_stack/contents/text_message_content.py 
14
15
16
17
18
19
20
21
22
23
24
@define
class TextMessageContent(BaseMessageContent):
    artifact: BaseArtifact = field(metadata={"serializable": True})

    @classmethod
    def from_deltas(cls, deltas: Sequence[BaseDeltaMessageContent]) -> TextMessageContent:
        text_deltas = [delta for delta in deltas if isinstance(delta, TextDeltaMessageContent)]

        artifact = TextArtifact(value="".join(delta.text for delta in text_deltas))

        return cls(artifact=artifact)

artifact = field(metadata={'serializable': True}) class-attribute instance-attribute

from_deltas(deltas) classmethod

Source code in griptape/common/prompt_stack/contents/text_message_content.py 
18
19
20
21
22
23
24
@classmethod
def from_deltas(cls, deltas: Sequence[BaseDeltaMessageContent]) -> TextMessageContent:
    text_deltas = [delta for delta in deltas if isinstance(delta, TextDeltaMessageContent)]

    artifact = TextArtifact(value="".join(delta.text for delta in text_deltas))

    return cls(artifact=artifact)

ToolAction

Bases: BaseAction

Represents an instance of an LLM using a Tool.

Attributes:

Name Type Description
tag str

The tag (unique identifier) of the action.
name str

The name (Tool name) of the action.
path Optional[str]

The path (Tool activity name) of the action.
input dict

The input (Tool params) of the action.
tool Optional[BaseTool]

The matched Tool of the action.
output Optional[BaseArtifact]

The output (Tool result) of the action.
Source code in griptape/common/actions/tool_action.py 
15
16
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
@define(kw_only=True)
class ToolAction(BaseAction):
    """Represents an instance of an LLM using a Tool.

    Attributes:
        tag: The tag (unique identifier) of the action.
        name: The name (Tool name) of the action.
        path: The path (Tool activity name) of the action.
        input: The input (Tool params) of the action.
        tool: The matched Tool of the action.
        output: The output (Tool result) of the action.
    """

    tag: str = field(metadata={"serializable": True})
    name: str = field(metadata={"serializable": True})
    path: Optional[str] = field(default=None, metadata={"serializable": True})
    input: dict = field(factory=dict, metadata={"serializable": True})
    tool: Optional[BaseTool] = field(default=None)
    output: Optional[BaseArtifact] = field(default=None)

    def __str__(self) -> str:
        return json.dumps(self.to_dict())

    def to_native_tool_name(self) -> str:
        parts = [self.name]

        if self.path is not None:
            parts.append(self.path)

        return "_".join(parts)

    @classmethod
    def from_native_tool_name(cls, native_tool_name: str) -> tuple[str, Optional[str]]:
        parts = native_tool_name.split("_", 1)

        if len(parts) == 1:
            name, path = parts[0], None
        else:
            name, path = parts

        return name, path

input = field(factory=dict, metadata={'serializable': True}) class-attribute instance-attribute

name = field(metadata={'serializable': True}) class-attribute instance-attribute

output = field(default=None) class-attribute instance-attribute

path = field(default=None, metadata={'serializable': True}) class-attribute instance-attribute

tag = field(metadata={'serializable': True}) class-attribute instance-attribute

tool = field(default=None) class-attribute instance-attribute

__str__()

Source code in griptape/common/actions/tool_action.py 
35
36
def __str__(self) -> str:
    return json.dumps(self.to_dict())

from_native_tool_name(native_tool_name) classmethod

Source code in griptape/common/actions/tool_action.py 
46
47
48
49
50
51
52
53
54
55
@classmethod
def from_native_tool_name(cls, native_tool_name: str) -> tuple[str, Optional[str]]:
    parts = native_tool_name.split("_", 1)

    if len(parts) == 1:
        name, path = parts[0], None
    else:
        name, path = parts

    return name, path

to_native_tool_name()

Source code in griptape/common/actions/tool_action.py 
38
39
40
41
42
43
44
def to_native_tool_name(self) -> str:
    parts = [self.name]

    if self.path is not None:
        parts.append(self.path)

    return "_".join(parts)