Conversation Memory

Conversation Memory#

Overview#

You can use Conversation Memory to give Gen AI Builder Structures the ability to keep track of the conversation across runs. All structures are created with AmazonDynamoDbConversationMemoryDriver by default.

Example#

from griptape.structures import Agent

agent = Agent()

agent.run("My favorite animal is a Liger.")
agent.run("What is my favorite animal?")

You can disable conversation memory in any structure by setting it to None :

from griptape.structures import Agent

Agent(conversation_memory=None)

Interaction With Structures#

Per Structure#

By default, Conversation Memory store(runs, metadata) are created for each run of the structure. Gen AI Builder takes the Structure’s base_audio_input_task ’s input and the output_task ’s output, storing them in the Run. Tasks that are neither the input task nor the output task are not stored in the Run.

from griptape.structures import Pipeline
from griptape.tasks import PromptTask

pipeline = Pipeline(
    tasks=[
        PromptTask("Respond to this request: {{ args[0] }}", id="input"),
        PromptTask("Improve the writing of this: {{ parent_output }}", id="improve"),
        PromptTask("Output this as a pirate: {{ parent_output }}", id="output"),
    ]
)

pipeline.run("My favorite animal is a Liger.")

if pipeline.conversation_memory is not None:
    for run in pipeline.conversation_memory.runs:
        print("Input", run.input.value)
        print("Output", run.output.value)
        print("\n\n")

In this example, the improve Task is “forgotten” after the Structure’s run is finished. This approach allows you to perform intermediary work within a Structure without it being stored in, and potentially cluttering, Conversation Memory.

Per Task#

You can change when Conversation Memory Runs are created by modifying Structure.conversation_memory_strategy from the default per_structure to per_task .

from griptape.structures import Pipeline
from griptape.tasks import PromptTask

pipeline = Pipeline(
    conversation_memory_strategy="per_task",
    tasks=[
        PromptTask("Respond to this request: {{ args[0] }}", id="input"),
        PromptTask("Improve the writing", id="improve"),
        PromptTask("Respond as a pirate", id="output"),
    ],
)

pipeline.run("My favorite animal is a Liger.")

if pipeline.conversation_memory is not None:
    for run in pipeline.conversation_memory.runs:
        print("Input", run.input.value)
        print("Output", run.output.value)
        print("\n\n")

Now, each Task creates a Conversation Memory Run when it runs. This eliminates the need to feed the output of Tasks into each other using context variables like {{ parent_output }} since the output of the previous Task is stored in Conversation Memory and loaded when the next Task runs.

To blend the two approaches, you can disable Conversation Memory on individual tasks by setting PromptTask.conversation_memory to None .

from griptape.structures import Pipeline
from griptape.tasks import PromptTask

pipeline = Pipeline(
    conversation_memory_strategy="per_task",
    tasks=[
        PromptTask("Respond to this request: {{ args[0] }}", id="input"),
        PromptTask("Improve the writing of this: {{ parent_output }}", id="improve", conversation_memory=None),
        PromptTask("Respond as a pirate", id="output"),
    ],
)

pipeline.run("My favorite animal is a Liger.")

if pipeline.conversation_memory is not None:
    for run in pipeline.conversation_memory.runs:
        print("Input", run.input.value)
        print("Output", run.output.value)
        print("\n\n")

Types of Memory#

Gen AI Builder provides several types of Conversation Memory to fit various use-cases.

Conversation Memory#

AmazonDynamoDbConversationMemoryDriver will keep track of the full task input and output for all runs.

from griptape.memory.structure import ConversationMemory
from griptape.structures import Agent

agent = Agent(conversation_memory=ConversationMemory())

agent.run("Hello!")

print(agent.conversation_memory)

You can set the max_runs parameter to limit how many runs are kept in memory.

from griptape.memory.structure import ConversationMemory
from griptape.structures import Agent

conversation_memory = ConversationMemory(max_runs=2)
agent = Agent(conversation_memory=conversation_memory)

agent.run("Run 1")
agent.run("Run 2")
agent.run("Run 3")
agent.run("Run 4")
agent.run("Run 5")

print(conversation_memory.runs[0].input == "run4")
print(conversation_memory.runs[1].input == "run5")

Summary Conversation Memory#

SummaryConversationMemory will progressively summarize task input and output of runs.

You can choose to offset which runs are summarized with the query_vector(vector, *, count=None, namespace=None, include_vectors=False, offset=None, **kwargs) parameter.

from griptape.memory.structure import SummaryConversationMemory
from griptape.structures import Agent
from griptape.utils import Conversation

agent = Agent(conversation_memory=SummaryConversationMemory(offset=2))

agent.run("Hello my name is John?")
agent.run("What is my name?")
agent.run("My favorite color is blue.")

print(Conversation(agent.conversation_memory))
Contents