pydantic_ai.conversation_store
Storing conversations as a log of messages, each stored once.
A ConversationStore keeps a conversation as a
chain of ConversationEntrys, each holding one
message and pointing at the entry before it, plus a
ConversationRecord that points at the newest
entry and carries what the messages can’t: usage, the deferred tool requests a paused run is
waiting on, and the scope the conversation belongs to.
Saving a conversation appends only the messages the store doesn’t have yet. When the history was rewritten instead of extended, by compaction for instance, the new messages branch off the last one both histories share. A fork points at an existing entry rather than copying the history before it.
One message in a conversation’s log, linked to the entry before it.
The entry’s identifier, minted when it is stored.
Not derived from the message: the same message can legitimately appear at two points in a history, and the entry before it is part of what makes an entry what it is.
Type: str Default: dataclasses.field(default_factory=(lambda: str(uuid7())))
The entry before this one, or None for the first entry in a history.
The message this entry holds.
Type: _messages.ModelMessage
What the entry holds, so other kinds of entries can join the log without a migration.
Type: Literal[‘message’] Default: 'message'
What a store keeps for a conversation besides its log: where its history ends, and what it carries.
The conversation this record is for.
Type: str
The newest entry in the conversation’s history, or None while it has no messages.
Usage accumulated across every run in the conversation.
Type: _usage.RunUsage Default: dataclasses.field(default_factory=(_usage.RunUsage))
The tool calls the conversation is waiting on, if its last run paused for them.
Type: DeferredToolRequests | None Default: None
Whose or where the conversation is, such as a user or a channel, to list conversations by.
Separate from conversation_id, which names the dialogue but says nothing about who it belongs to.
Type: str | None Default: None
The conversation_id of the conversation a fork copied this one from.
Type: str | None Default: None
The last entry of the history the fork kept, or None if it kept none or this conversation isn’t a fork.
Type: str | None Default: None
The last entries of the branches this conversation has left, oldest first.
branch adds the history it leaves, as
does a save that rewrites the history, so an app can list them and
branch back to one with from_entry.
Type: list[str] Default: dataclasses.field(default_factory=(list[str]))
A token that changes on every save, so concurrent writers can’t overwrite each other unnoticed.
Opaque: compare it for equality only. A conversation deleted and created again under the same ID gets a new one, so a copy loaded before the delete can’t be saved over it.
Type: str Default: dataclasses.field(default_factory=(lambda: str(uuid7())))
When the conversation was last saved.
Type: datetime Default: dataclasses.field(default_factory=now_utc)
The storage format the record and its entries were written in.
Type: Literal[1] Default: 1
Bases: UserError
Raised when a conversation changed in the store since it was loaded, so saving it would lose those changes.
It’s an expected outcome when more than one writer saves the same conversation: catch it, load the
conversation again, and retry. It subclasses UserError, so a
broad except UserError catches it too.
The conversation that changed.
Type: str Default: conversation_id
Bases: ABC
A store for conversations, keeping each one’s messages as a log in which each message is stored once.
load and
save a
Conversation around each run:
conversation = await store.load(conversation_id) or Conversation(conversation_id=conversation_id)
result = await agent.run(prompt, conversation=conversation)
await store.save(result)
The stored conversation only ever grows: a save appends the messages a run added, even when a
capability trimmed or rewrote the history that run was sent. To regenerate or edit an earlier
turn, branch it from an earlier
message: the history it leaves stays stored, so it can branch back. To explore from a point
without moving the conversation, fork
a new one from it.
Pass scope= to the methods that read or change a stored conversation to keep it to its owner: a
conversation saved with another scope is treated as one the store doesn’t have.
A backend implements the five storage primitives below; the store works out which messages are new, where a history branches, and what a fork shares.
@async
def load(
conversation_id: str,
*,
limit: int | None = None,
scope: str | None = None,
) -> Conversation | None
Load a conversation, or None if the store doesn’t have it.
conversation_id : str
The conversation to load.
Only load it if it was saved with this scope.
Load only its last limit messages, for a run that only needs the recent turns.
Saving the result still appends to the whole stored conversation. The window can
start in the middle of a turn, say with a tool result whose call it left out; a
request repairs such a history before it’s sent.
UserError— Iflimitis negative.
@async
def load_entries(
conversation_id: str,
*,
limit: int | None = None,
scope: str | None = None,
) -> list[ConversationEntry] | None
Load a conversation’s history with each message’s entry, or None if the store doesn’t have it.
An entry’s id identifies its message
in the store, so it’s what to tie a rating or a billing row to. save returns the entries it
stored.
list[ConversationEntry] | None
conversation_id : str
The conversation to load.
Load only its last limit messages.
Only load it if it was saved with this scope.
UserError— Iflimitis negative.
@async
def save(
conversation: Conversation | AgentRunResult[Any],
*,
scope: str | None = None,
) -> list[ConversationEntry]
Save a conversation, storing only the messages the store doesn’t have yet.
When the conversation knows its
new_message_index, as one that
was loaded, or a run’s result from one, does, the messages from that index on are appended
to the stored conversation. The messages before it aren’t compared while they’re the ones it
was set after, so a history a capability trimmed or rewrote doesn’t change what’s stored.
Otherwise the messages are compared with the stored history: with the same number of stored
messages, up to the last one, as new_message_index says it holds, say after the
conversation was serialized, or with all of them when it’s None. Those that extend it are
appended, and if it was rewritten, the messages from the first difference on branch off the
last message both share, keeping the old branch stored. Messages are matched with ==, so a
change only to a field message equality ignores, like a file’s media_type or identifier,
isn’t stored.
The save is checked against the conversation’s
revision: it must be the stored one, or
None for a conversation the store doesn’t have yet. On success, revision is advanced on
the conversation passed in, so it can be saved again after another run, and so is its
new_message_index.
list[ConversationEntry] — The entries this save stored, in order, so their
list[ConversationEntry] — ids can be recorded next to the
list[ConversationEntry] — messages, say with a billing row or a rating. Empty when there was nothing new.
conversation : Conversation | AgentRunResult[Any]
The conversation to save, or a run’s result to save its
conversation. A saved result’s
conversation then carries the new revision, so it can be continued or saved again.
Whose or where the conversation is, to list it by. A conversation the store has must have been saved with it, and one it doesn’t have is saved with it. Defaults to the scope it was saved with before.
ConversationConflict— If the conversation was saved or deleted since it was loaded, one with itsconversation_idalready exists when it wasn’t loaded from the store, or it was saved with anotherscope.
@async
def branch(
conversation_id: str,
*,
at: int | None = None,
from_entry: str | None = None,
expected_revision: str | None,
scope: str | None = None,
) -> Conversation
Continue a conversation from an earlier point, to regenerate or edit what came after it, or switch branches.
The history it leaves stays stored, and is added to the record’s
branches, so the conversation can
branch back to it with from_entry. Continue the returned conversation and save it as usual: the
new turn goes on the branch.
conversation_id : str
The conversation to branch.
How many of its current messages to keep.
The entry to continue from instead: one in its current history, or in a branch it left.
The revision the
conversation was loaded with, so a branch can’t undo a turn someone else saved since.
Only branch it if it was saved with this scope.
UserError— If the store doesn’t have the conversation,atis out of range,from_entryisn’t in its history or a branch it left, or not exactly one ofatandfrom_entryis given.ConversationConflict— If the conversation was saved or deleted since it was loaded.
@async
def fork(
conversation_id: str,
*,
at: int | None = None,
from_entry: str | None = None,
new_conversation_id: str | None = None,
scope: str | None = None,
source_scope: str | None = None,
) -> Conversation
Start a new conversation from a stored one’s history, sharing its entries rather than copying them.
Fork the same point more than once to explore alternatives side by side, while the original conversation carries on.
conversation_id : str
The conversation to fork.
How many of its messages the fork keeps. Defaults to all of them.
The entry to fork from instead: one in its history, or in a branch it left.
The fork’s identifier. Defaults to a new one.
The fork’s scope. Defaults to the forked conversation’s.
Only fork the conversation if it was saved with this scope.
UserError— If the store doesn’t have the conversation,atis out of range,from_entryisn’t in its history or a branch it left, or bothatandfrom_entryare given.ConversationConflict— If a conversation withnew_conversation_idis already stored.
@async
def list_conversations(
*,
scope: str | None = None,
limit: int | None = None,
) -> list[ConversationRecord]
The stored conversations, most recently saved first.
list[ConversationRecord]
Only list the conversations saved with this scope.
List at most this many, the most recently saved.
UserError— Iflimitis negative.
@async
def delete(conversation_id: str, *, scope: str | None = None) -> None
Delete a conversation, so it no longer loads or lists.
With scope, a conversation saved with another scope is left as it is.
Whether the entries no other conversation’s history uses are deleted with it depends on the
backend: InMemoryConversationStore
deletes them, and each backend documents what it does. Don’t rely on delete alone to
erase a conversation’s messages unless your backend says it does.
@async
def read_history(
head_id: str | None,
*,
limit: int | None = None,
) -> list[ConversationEntry]
The entries from the first one up to head_id, in order, or only the last limit of them.
list[ConversationEntry]
@abstractmethod
@async
def read_record(conversation_id: str) -> ConversationRecord | None
The record for a conversation, or None if the store doesn’t have it.
ConversationRecord | None
@abstractmethod
@async
def read_records(
*,
scope: str | None = None,
limit: int | None = None,
) -> Sequence[ConversationRecord]
The stored records, or those with the given scope, newest updated_at first, and at most limit of them.
Sequence[ConversationRecord]
@abstractmethod
@async
def write_record(
record: ConversationRecord,
*,
new_entries: Sequence[ConversationEntry] = (),
expected_revision: str | None,
) -> None
Store a record and the new entries its history adds, replacing the record with expected_revision.
new_entries are in order, each after the one it points at. A backend that can should
store them and the record together, so a conflict leaves neither behind.
ConversationConflict— If the stored record’s revision isn’texpected_revision, a record exists whenexpected_revisionisNone, orrecord.head_idnames an entry that neither the store nornew_entrieshas, because a concurrent delete or rewrite of the conversation a fork points into released it.
@abstractmethod
@async
def delete_record(conversation_id: str, *, expected_revision: str) -> None
Delete a conversation’s record, and, where the backend can do so safely, the entries no remaining record’s history reaches.
A backend that keeps those entries, say because it can’t tell them from ones a concurrent save is about to use, should say so in its documentation.
ConversationConflict— If no record withexpected_revisionis stored, because the conversation was saved, replaced or deleted since it was read.
@abstractmethod
@async
def read_entry(entry_id: str) -> ConversationEntry
The entry with this identifier.
Backends that can read a whole history in one query should override
read_history instead of
relying on one call per entry.
ConversationEntry
Bases: ConversationStore
A ConversationStore that keeps everything in memory, for tests and examples.
It keeps its own copy of each record and message, as a store that serializes them would, so changing the conversation or record it was given or returned doesn’t change what was saved.