@define
class SummaryConversationMemory(BaseConversationMemory):
"""Conversation memory that automatically summarizes older runs, keeping a configurable number of recent runs in full detail.
The memory stores **all** runs in ``self.runs`` and automatically generates an LLM-powered
summary of older runs as new ones are added. Only the summary and the most recent runs
(controlled by ``offset``) are included in the prompt context via ``to_prompt_stack()``.
Note on display utilities:
- **Conversation utility** (``griptape.utils.Conversation``): Displays **all** runs stored
in memory, not just the summary or the unsummarized portion. When used with
``SummaryConversationMemory``, it prints every Q/A pair from ``self.runs`` followed
by the generated summary. This is expected behavior -- the runs list preserves the
full history for inspection, while the summary is used internally for prompt context.
- **Chat utility** (``griptape.utils.Chat``): Calls ``Structure.run()``, which invokes
``to_prompt_stack()`` internally. This means only the summary and the unsummarized
recent runs (those within ``offset``) are sent to the LLM as context.
Attributes:
offset: Maximum number of recent runs to keep unsummarized. When a new run is
added and the count of unsummarized runs exceeds ``offset``, the oldest excess runs are summarized
into a single condensed summary string. Defaults to 1.
autoprune: Inherited from ``BaseConversationMemory``. When enabled,
``add_to_prompt_stack()`` further trims the prompt context to fit within
the model's token limit, on top of the summary/offset pruning already
performed by ``to_prompt_stack()``. Does not remove runs from ``self.runs``.
"""
offset: int = field(default=1, kw_only=True, metadata={"serializable": True})
prompt_driver: BasePromptDriver = field(
kw_only=True, default=Factory(lambda: Defaults.drivers_config.prompt_driver)
)
summary: str | None = field(default=None, kw_only=True, metadata={"serializable": True})
summary_index: int = field(default=0, kw_only=True, metadata={"serializable": True})
summary_get_template: J2 = field(default=Factory(lambda: J2("memory/conversation/summary.j2")), kw_only=True)
summarize_conversation_get_template: J2 = field(
default=Factory(lambda: J2("memory/conversation/summarize_conversation.j2")),
kw_only=True,
)
# Set meta['summary'] after initializing self.summary, because load_runs() will overwrite it with an empty value from meta.
def __attrs_post_init__(self) -> None:
if self.summary is not None:
self.meta["summary"] = self.summary
self.meta["summary_index"] = self.summary_index
super().__attrs_post_init__()
def to_prompt_stack(self, last_n: int | None = None) -> PromptStack:
stack = PromptStack()
if self.summary:
stack.add_user_message(self.summary_get_template.render(summary=self.summary))
for r in self.unsummarized_runs(last_n):
stack.add_user_message(r.input)
stack.add_assistant_message(r.output)
return stack
def unsummarized_runs(self, last_n: int | None = 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
return summary_index_runs
return summary_index_runs
def try_add_run(self, run: Run) -> None:
self.runs.append(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 | None, runs: list[Run]) -> str | None:
try:
if len(runs) > 0:
summary = self.summarize_conversation_get_template.render(summary=previous_summary, runs=runs)
return self.prompt_driver.run(
PromptStack(messages=[Message(summary, role=Message.USER_ROLE)]),
).to_text()
return previous_summary
except Exception as e:
logging.exception("Error summarizing memory: %s(%s)", type(e).__name__, e)
return previous_summary
def load_runs(self) -> list[Run]:
runs = super().load_runs()
self.summary = self.meta.get("summary")
self.summary_index = self.meta.get("summary_index", 0)
return runs