Skip to content

Structure

__all__ = ['Run', 'ConversationMemory', 'SummaryConversationMemory'] module-attribute

ConversationMemory

Source code in griptape/griptape/memory/structure/conversation_memory.py 
13
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
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
@define
class ConversationMemory:
    type: str = field(default=Factory(lambda self: self.__class__.__name__, takes_self=True), kw_only=True)
    driver: Optional[BaseConversationMemoryDriver] = field(default=None, kw_only=True)
    runs: list[Run] = field(factory=list, kw_only=True)
    structure: Structure = field(init=False)
    autoload: bool = field(default=True, kw_only=True)
    autoprune: bool = field(default=True, kw_only=True)
    max_runs: Optional[int] = field(default=None, kw_only=True)

    def __attrs_post_init__(self) -> None:
        if self.driver and self.autoload:
            memory = self.driver.load()
            if memory is not None:
                [self.add_run(r) for r in memory.runs]

    def add_run(self, run: Run) -> ConversationMemory:
        self.before_add_run()
        self.try_add_run(run)
        self.after_add_run()

        return self

    def before_add_run(self) -> None:
        pass

    def try_add_run(self, run: Run) -> None:
        self.runs.append(run)

        if self.max_runs:
            while len(self.runs) > self.max_runs:
                self.runs.pop(0)

    def after_add_run(self) -> None:
        if self.driver:
            self.driver.store(self)

    def to_json(self) -> str:
        return json.dumps(self.to_dict(), indent=2)

    def to_dict(self) -> dict:
        from griptape.schemas import ConversationMemorySchema

        return dict(ConversationMemorySchema().dump(self))

    def to_prompt_stack(self, last_n: Optional[int] = None) -> PromptStack:
        prompt_stack = PromptStack()
        runs = self.runs[-last_n:] if last_n else self.runs
        for run in runs:
            prompt_stack.add_user_input(run.input)
            prompt_stack.add_assistant_input(run.output)
        return prompt_stack

    @classmethod
    def from_dict(cls, memory_dict: dict) -> ConversationMemory:
        from griptape.schemas import ConversationMemorySchema

        return ConversationMemorySchema().load(memory_dict)

    @classmethod
    def from_json(cls, memory_json: str) -> ConversationMemory:
        return ConversationMemory.from_dict(json.loads(memory_json))

autoload: bool = field(default=True, kw_only=True) class-attribute instance-attribute

autoprune: bool = field(default=True, kw_only=True) class-attribute instance-attribute

driver: Optional[BaseConversationMemoryDriver] = field(default=None, kw_only=True) class-attribute instance-attribute

max_runs: Optional[int] = field(default=None, kw_only=True) class-attribute instance-attribute

runs: list[Run] = field(factory=list, kw_only=True) class-attribute instance-attribute

structure: Structure = field(init=False) class-attribute instance-attribute

type: str = field(default=Factory(lambda : self.__class__.__name__, takes_self=True), kw_only=True) class-attribute instance-attribute

__attrs_post_init__()

Source code in griptape/griptape/memory/structure/conversation_memory.py 
23
24
25
26
27
def __attrs_post_init__(self) -> None:
    if self.driver and self.autoload:
        memory = self.driver.load()
        if memory is not None:
            [self.add_run(r) for r in memory.runs]

add_run(run)

Source code in griptape/griptape/memory/structure/conversation_memory.py 
29
30
31
32
33
34
def add_run(self, run: Run) -> ConversationMemory:
    self.before_add_run()
    self.try_add_run(run)
    self.after_add_run()

    return self

after_add_run()

Source code in griptape/griptape/memory/structure/conversation_memory.py 
46
47
48
def after_add_run(self) -> None:
    if self.driver:
        self.driver.store(self)

before_add_run()

Source code in griptape/griptape/memory/structure/conversation_memory.py 
36
37
def before_add_run(self) -> None:
    pass

from_dict(memory_dict) classmethod

Source code in griptape/griptape/memory/structure/conversation_memory.py 
66
67
68
69
70
@classmethod
def from_dict(cls, memory_dict: dict) -> ConversationMemory:
    from griptape.schemas import ConversationMemorySchema

    return ConversationMemorySchema().load(memory_dict)

from_json(memory_json) classmethod

Source code in griptape/griptape/memory/structure/conversation_memory.py 
72
73
74
@classmethod
def from_json(cls, memory_json: str) -> ConversationMemory:
    return ConversationMemory.from_dict(json.loads(memory_json))

to_dict()

Source code in griptape/griptape/memory/structure/conversation_memory.py 
53
54
55
56
def to_dict(self) -> dict:
    from griptape.schemas import ConversationMemorySchema

    return dict(ConversationMemorySchema().dump(self))

to_json()

Source code in griptape/griptape/memory/structure/conversation_memory.py 
50
51
def to_json(self) -> str:
    return json.dumps(self.to_dict(), indent=2)

to_prompt_stack(last_n=None)

Source code in griptape/griptape/memory/structure/conversation_memory.py 
58
59
60
61
62
63
64
def to_prompt_stack(self, last_n: Optional[int] = None) -> PromptStack:
    prompt_stack = PromptStack()
    runs = self.runs[-last_n:] if last_n else self.runs
    for run in runs:
        prompt_stack.add_user_input(run.input)
        prompt_stack.add_assistant_input(run.output)
    return prompt_stack

try_add_run(run)

Source code in griptape/griptape/memory/structure/conversation_memory.py 
39
40
41
42
43
44
def try_add_run(self, run: Run) -> None:
    self.runs.append(run)

    if self.max_runs:
        while len(self.runs) > self.max_runs:
            self.runs.pop(0)

Run

Source code in griptape/griptape/memory/structure/run.py 
5
6
7
8
9
@define
class Run:
    id: str = field(default=Factory(lambda: uuid.uuid4().hex), kw_only=True)
    input: str = field(kw_only=True)
    output: str = field(kw_only=True)

id: str = field(default=Factory(lambda : uuid.uuid4().hex), kw_only=True) class-attribute instance-attribute

input: str = field(kw_only=True) class-attribute instance-attribute

output: str = field(kw_only=True) class-attribute instance-attribute

SummaryConversationMemory

Bases: ConversationMemory

Source code in griptape/griptape/memory/structure/summary_conversation_memory.py 
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
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
@define
class SummaryConversationMemory(ConversationMemory):
    offset: int = field(default=1, kw_only=True)
    prompt_driver: BasePromptDriver = field(
        default=Factory(lambda: OpenAiChatPromptDriver(model=OpenAiTokenizer.DEFAULT_OPENAI_GPT_3_CHAT_MODEL)),
        kw_only=True,
    )
    summary: Optional[str] = field(default=None, kw_only=True)
    summary_index: int = field(default=0, kw_only=True)
    summary_template_generator: J2 = field(default=Factory(lambda: J2("memory/conversation/summary.j2")), kw_only=True)
    summarize_conversation_template_generator: J2 = field(
        default=Factory(lambda: J2("memory/conversation/summarize_conversation.j2")), kw_only=True
    )

    @classmethod
    def from_dict(cls, memory_dict: dict) -> SummaryConversationMemory:
        return SummaryConversationMemorySchema().load(memory_dict)

    @classmethod
    def from_json(cls, memory_json: str) -> SummaryConversationMemory:
        return SummaryConversationMemory.from_dict(json.loads(memory_json))

    def to_prompt_stack(self, last_n: Optional[int] = None) -> PromptStack:
        stack = PromptStack()
        if self.summary:
            stack.add_user_input(self.summary_template_generator.render(summary=self.summary))

        for r in self.unsummarized_runs(last_n):
            stack.add_user_input(r.input)
            stack.add_assistant_input(r.output)

        return stack

    def to_dict(self) -> dict:
        return dict(SummaryConversationMemorySchema().dump(self))

    def unsummarized_runs(self, last_n: Optional[int] = None) -> list[Run]:
        summary_index_runs = self.runs[self.summary_index :]

        if last_n:
            last_n_runs = self.runs[-last_n:]

            if len(summary_index_runs) > len(last_n_runs):
                return last_n_runs
            else:
                return summary_index_runs
        else:
            return summary_index_runs

    def try_add_run(self, run: Run) -> None:
        super().try_add_run(run)

        unsummarized_runs = self.unsummarized_runs()
        runs_to_summarize = unsummarized_runs[: max(0, len(unsummarized_runs) - self.offset)]

        if len(runs_to_summarize) > 0:
            self.summary = self.summarize_runs(self.summary, runs_to_summarize)
            self.summary_index = 1 + self.runs.index(runs_to_summarize[-1])

    def summarize_runs(self, previous_summary: str, runs: list[Run]) -> str:
        try:
            if len(runs) > 0:
                summary = self.summarize_conversation_template_generator.render(summary=previous_summary, runs=runs)
                return self.prompt_driver.run(
                    prompt_stack=PromptStack(inputs=[PromptStack.Input(summary, role=PromptStack.USER_ROLE)])
                ).to_text()
            else:
                return previous_summary
        except Exception as e:
            logging.error(f"Error summarizing memory: {type(e).__name__}({e})")

            return previous_summary

offset: int = field(default=1, kw_only=True) class-attribute instance-attribute

prompt_driver: BasePromptDriver = field(default=Factory(lambda : OpenAiChatPromptDriver(model=OpenAiTokenizer.DEFAULT_OPENAI_GPT_3_CHAT_MODEL)), kw_only=True) class-attribute instance-attribute

summarize_conversation_template_generator: J2 = field(default=Factory(lambda : J2('memory/conversation/summarize_conversation.j2')), kw_only=True) class-attribute instance-attribute

summary: Optional[str] = field(default=None, kw_only=True) class-attribute instance-attribute

summary_index: int = field(default=0, kw_only=True) class-attribute instance-attribute

summary_template_generator: J2 = field(default=Factory(lambda : J2('memory/conversation/summary.j2')), kw_only=True) class-attribute instance-attribute

from_dict(memory_dict) classmethod

Source code in griptape/griptape/memory/structure/summary_conversation_memory.py 
32
33
34
@classmethod
def from_dict(cls, memory_dict: dict) -> SummaryConversationMemory:
    return SummaryConversationMemorySchema().load(memory_dict)

from_json(memory_json) classmethod

Source code in griptape/griptape/memory/structure/summary_conversation_memory.py 
36
37
38
@classmethod
def from_json(cls, memory_json: str) -> SummaryConversationMemory:
    return SummaryConversationMemory.from_dict(json.loads(memory_json))

summarize_runs(previous_summary, runs)

Source code in griptape/griptape/memory/structure/summary_conversation_memory.py 
77
78
79
80
81
82
83
84
85
86
87
88
89
def summarize_runs(self, previous_summary: str, runs: list[Run]) -> str:
    try:
        if len(runs) > 0:
            summary = self.summarize_conversation_template_generator.render(summary=previous_summary, runs=runs)
            return self.prompt_driver.run(
                prompt_stack=PromptStack(inputs=[PromptStack.Input(summary, role=PromptStack.USER_ROLE)])
            ).to_text()
        else:
            return previous_summary
    except Exception as e:
        logging.error(f"Error summarizing memory: {type(e).__name__}({e})")

        return previous_summary

to_dict()

Source code in griptape/griptape/memory/structure/summary_conversation_memory.py 
51
52
def to_dict(self) -> dict:
    return dict(SummaryConversationMemorySchema().dump(self))

to_prompt_stack(last_n=None)

Source code in griptape/griptape/memory/structure/summary_conversation_memory.py 
40
41
42
43
44
45
46
47
48
49
def to_prompt_stack(self, last_n: Optional[int] = None) -> PromptStack:
    stack = PromptStack()
    if self.summary:
        stack.add_user_input(self.summary_template_generator.render(summary=self.summary))

    for r in self.unsummarized_runs(last_n):
        stack.add_user_input(r.input)
        stack.add_assistant_input(r.output)

    return stack

try_add_run(run)

Source code in griptape/griptape/memory/structure/summary_conversation_memory.py 
67
68
69
70
71
72
73
74
75
def try_add_run(self, run: Run) -> None:
    super().try_add_run(run)

    unsummarized_runs = self.unsummarized_runs()
    runs_to_summarize = unsummarized_runs[: max(0, len(unsummarized_runs) - self.offset)]

    if len(runs_to_summarize) > 0:
        self.summary = self.summarize_runs(self.summary, runs_to_summarize)
        self.summary_index = 1 + self.runs.index(runs_to_summarize[-1])

unsummarized_runs(last_n=None)

Source code in griptape/griptape/memory/structure/summary_conversation_memory.py 
54
55
56
57
58
59
60
61
62
63
64
65
def unsummarized_runs(self, last_n: Optional[int] = None) -> list[Run]:
    summary_index_runs = self.runs[self.summary_index :]

    if last_n:
        last_n_runs = self.runs[-last_n:]

        if len(summary_index_runs) > len(last_n_runs):
            return last_n_runs
        else:
            return summary_index_runs
    else:
        return summary_index_runs