fastrag.query.chat#

Chat persistence: sessions, chats, and messages.

The ChatStore contract and its PostgreSQL implementation. Storage only — no FastAPI, no HTTP types, no SSE; a serving layer built on top of a ChatStore owns all of that. See fastrag.query.chat.base for what a serving layer built on this package gets: session create/read/touch, chat create/list/read/delete, and message append/list/update-last-assistant.

class fastrag.query.chat.Chat(id, session_id, created_at, expires_at)[source]#

Bases: object

One conversation: an ordered, linear list of messages.

Parameters:
id#

ULID, minted on creation.

Type:

str

session_id#

The owning Session.

Type:

str

created_at#

Derived from id.

Type:

datetime.datetime

expires_at#

When this chat (and, by cascade, its messages) stops being usable. Independent of its session’s expires_at — a chat can go idle while its session stays active.

Type:

datetime.datetime

class fastrag.query.chat.ChatMessage(id, chat_id, created_at, role, content=None, sources=None, finish_reason=None, error_code=None, error_message=None)[source]#

Bases: object

One persisted turn, from user, assistant, or system.

A different thing from Message, the model-facing turn a ChatModel consumes — see to_llm_history() for the conversion.

Parameters:

  • id (str)

  • chat_id (str)

  • created_at (datetime)

  • role (Literal['user', 'assistant', 'system'])

  • content (str | None)

  • sources (list[dict[str, Any]] | None)

  • finish_reason (Literal['stop', 'length', 'error', 'interrupted'] | None)

  • error_code (Literal['MODEL_ERROR', 'TIMEOUT', 'RATE_LIMIT', 'CONTEXT_TOO_LONG', 'NETWORK_ERROR', 'SERVICE_UNAVAILABLE', 'INVALID_REQUEST', 'INVALID_RESPONSE', 'INTERNAL_ERROR'] | None)

  • error_message (str | None)

id#

ULID, minted on creation.

Type:

str

chat_id#

The owning Chat.

Type:

str

created_at#

Derived from id.

Type:

datetime.datetime

role#

Who sent the turn.

Type:

Literal[‘user’, ‘assistant’, ‘system’]

content#

The turn’s text. None only on an assistant turn whose finish_reason is "error" and nothing was generated before it failed.

Type:

str | None

sources#

The documents the answer was grounded in, serialized as plain JSON objects (a store never interprets their shape). None for a user/system turn.

Type:

list[dict[str, Any]] | None

finish_reason#

Why an assistant turn stopped. Required for assistant turns, absent otherwise.

Type:

Literal[‘stop’, ‘length’, ‘error’, ‘interrupted’] | None

error_code#

Set exactly when finish_reason == "error".

Type:

Literal[‘MODEL_ERROR’, ‘TIMEOUT’, ‘RATE_LIMIT’, ‘CONTEXT_TOO_LONG’, ‘NETWORK_ERROR’, ‘SERVICE_UNAVAILABLE’, ‘INVALID_REQUEST’, ‘INVALID_RESPONSE’, ‘INTERNAL_ERROR’] | None

error_message#

Set exactly when finish_reason == "error".

Type:

str | None

class fastrag.query.chat.ChatStore[source]#

Bases: AbstractAsyncContextManager[ChatStore], ABC

Pluggable abstract persistence for sessions, chats, and messages.

Storage only: no FastAPI, no HTTP types, no SSE, no QueryEngine — a serving layer built on top of a ChatStore owns all of that.

abstractmethod async create_session(*, ttl_days, ip_address=None, country=None)[source]#

Mint and persist a new session.

Parameters:

  • ttl_days (int) – Days until the session expires without activity.

  • ip_address (str | None) – The creating request’s peer address, if known.

  • country (str | None) – A two-letter country code, if known.

Return type:

Session

abstractmethod async get_session(session_id)[source]#

Fetch an active session by id.

Returns:

The session, or None if session_id is unknown or its expires_at has already passed.

Parameters:

session_id (str)

Return type:

Session | None

abstractmethod async touch_session(session_id, *, ttl_days, min_bump_days)[source]#

Bump an active session’s expiry, throttled.

The expiry is only written when the new value is at least min_bump_days later than the stored one, so an active conversation stays alive without a write on every request.

Parameters:

  • session_id (str) – The session to touch.

  • ttl_days (int) – Days from now the session’s new expiry would be, if bumped.

  • min_bump_days (int) – Minimum headroom, in days, the candidate expiry must gain over the stored one before it is written.

Returns:

The session as it now stands (bumped or not), or None if session_id is unknown or already expired.

Return type:

Session | None

abstractmethod async create_chat(*, session_id, ttl_days)[source]#

Start a new, empty chat owned by session_id.

Parameters:

  • session_id (str) – The owning session.

  • ttl_days (int) – Days until the chat expires without activity.

Return type:

Chat

abstractmethod async get_chat(chat_id, *, session_id)[source]#

Fetch a chat, scoped to its owning session.

Returns:

The chat, or None if chat_id does not exist or does not belong to session_id.

Parameters:

  • chat_id (str)

  • session_id (str)

Return type:

Chat | None

abstractmethod async list_chats(session_id, *, before=None, limit=50)[source]#

List session_id’s chats, newest-first.

Parameters:

  • session_id (str) – The owning session.

  • before (str | None) – When given, only chats older than this chat id are returned — pass a previous page’s last item’s id to continue paging.

  • limit (int) – Maximum chats to return.

Return type:

Page[Chat]

abstractmethod async delete_chat(chat_id, *, session_id)[source]#

Delete a chat and every message in it.

Parameters:

  • chat_id (str) – The chat to delete.

  • session_id (str) – The owning session — chat_id is only deleted if it belongs to this session.

Returns:

True if a chat was deleted, False if chat_id did not exist or did not belong to session_id.

Return type:

bool

abstractmethod async append_message(chat_id, *, role, content=None, sources=None, finish_reason=None, error_code=None, error_message=None)[source]#

Append a new turn to a chat.

Parameters:

  • chat_id (str) – The chat this turn belongs to.

  • role (Literal['user', 'assistant', 'system']) – Who sent the turn.

  • content (str | None) – The turn’s text. May be None only for an assistant turn whose finish_reason is "error".

  • sources (list[dict[str, Any]] | None) – The documents the answer was grounded in, as plain JSON objects. Only meaningful for an assistant turn.

  • finish_reason (Literal['stop', 'length', 'error', 'interrupted'] | None) – Required for an assistant turn, must be absent otherwise.

  • error_code (Literal['MODEL_ERROR', 'TIMEOUT', 'RATE_LIMIT', 'CONTEXT_TOO_LONG', 'NETWORK_ERROR', 'SERVICE_UNAVAILABLE', 'INVALID_REQUEST', 'INVALID_RESPONSE', 'INTERNAL_ERROR'] | None) – Set exactly when finish_reason == "error".

  • error_message (str | None) – Set exactly when finish_reason == "error".

Return type:

ChatMessage

abstractmethod async list_messages(chat_id, *, before=None, limit=50)[source]#

List a chat’s messages, newest-first.

Parameters:

  • chat_id (str) – The chat to list.

  • before (str | None) – When given, only messages older than this message id are returned — pass a previous page’s last item’s id to continue paging.

  • limit (int) – Maximum messages to return.

Return type:

Page[ChatMessage]

abstractmethod async update_last_assistant_message(chat_id, *, content, sources=None, finish_reason, error_code=None, error_message=None)[source]#

Overwrite a chat’s most recent assistant turn in place.

The storage primitive behind both the /retry flow (regenerate the last answer) and the /edit flow (regenerate after editing an earlier turn): both end by replacing the chat’s latest assistant message with a new outcome, in place, rather than deleting and re-appending.

Parameters:

  • chat_id (str) – The chat whose latest assistant message is updated.

  • content (str | None) – The turn’s new text. May be None only when finish_reason == "error".

  • sources (list[dict[str, Any]] | None) – The documents the new answer was grounded in.

  • finish_reason (Literal['stop', 'length', 'error', 'interrupted']) – Why generation stopped this time.

  • error_code (Literal['MODEL_ERROR', 'TIMEOUT', 'RATE_LIMIT', 'CONTEXT_TOO_LONG', 'NETWORK_ERROR', 'SERVICE_UNAVAILABLE', 'INVALID_REQUEST', 'INVALID_RESPONSE', 'INTERNAL_ERROR'] | None) – Set exactly when finish_reason == "error".

  • error_message (str | None) – Set exactly when finish_reason == "error".

Returns:

The updated message, or None if chat_id has no assistant message yet.

Return type:

ChatMessage | None

class fastrag.query.chat.Page(items, has_more)[source]#

Bases: Generic[T]

One page of a ULID-cursor listing.

Parameters:
items#

This page’s rows, newest-first (highest id first).

Type:

list[fastrag.query.chat.base.T]

has_more#

Whether an older row exists beyond this page — pass this page’s last item’s id as the next call’s before to fetch it.

Type:

bool

class fastrag.query.chat.PostgresChatStore(*, database_url, schema='fastrag_chat', pool_size=5)[source]#

Bases: ChatStore

Persist sessions, chats, and messages on a PostgreSQL server.

async with PostgresChatStore(database_url=DATABASE_URL) as chat_store:
    session = await chat_store.create_session(ttl_days=30)
Parameters:

  • database_url (str) – postgresql://user:password@host:port/database. Accepted as a plain postgresql:// URL; the asyncpg driver is selected internally.

  • schema (str) – The Postgres schema this store’s tables live in. Created if absent. Defaults to a dedicated schema so the store never collides with application tables.

  • pool_size (int) – Connections held by this store’s engine.

property engine: AsyncEngine#

The store’s underlying async engine.

The serving layer (fastrag.serve) uses this engine — not one it owns — for two Postgres-specific things that need a raw engine rather than a ChatStore method: the per-chat advisory lock that serializes generations, and the /healthz connection check. Requires the store to have been entered (__aenter__), else it raises fastrag.exceptions.ChatStoreError. Not part of the ChatStore contract — a non-Postgres store may not offer one (in which case those two serving features are unavailable).

async create_session(*, ttl_days, ip_address=None, country=None)[source]#

See create_session().

Parameters:

  • ttl_days (int)

  • ip_address (str | None)

  • country (str | None)

Return type:

Session

async get_session(session_id)[source]#

See get_session().

Parameters:

session_id (str)

Return type:

Session | None

async touch_session(session_id, *, ttl_days, min_bump_days)[source]#

See touch_session().

Parameters:

  • session_id (str)

  • ttl_days (int)

  • min_bump_days (int)

Return type:

Session | None

async create_chat(*, session_id, ttl_days)[source]#

See create_chat().

Parameters:

  • session_id (str)

  • ttl_days (int)

Return type:

Chat

async get_chat(chat_id, *, session_id)[source]#

See get_chat().

Parameters:

  • chat_id (str)

  • session_id (str)

Return type:

Chat | None

async list_chats(session_id, *, before=None, limit=50)[source]#

See list_chats().

Parameters:

  • session_id (str)

  • before (str | None)

  • limit (int)

Return type:

Page[Chat]

async delete_chat(chat_id, *, session_id)[source]#

See delete_chat().

Parameters:

  • chat_id (str)

  • session_id (str)

Return type:

bool

async append_message(chat_id, *, role, content=None, sources=None, finish_reason=None, error_code=None, error_message=None)[source]#

See append_message().

Parameters:

  • chat_id (str)

  • role (Literal['user', 'assistant', 'system'])

  • content (str | None)

  • sources (list[dict[str, Any]] | None)

  • finish_reason (Literal['stop', 'length', 'error', 'interrupted'] | None)

  • error_code (Literal['MODEL_ERROR', 'TIMEOUT', 'RATE_LIMIT', 'CONTEXT_TOO_LONG', 'NETWORK_ERROR', 'SERVICE_UNAVAILABLE', 'INVALID_REQUEST', 'INVALID_RESPONSE', 'INTERNAL_ERROR'] | None)

  • error_message (str | None)

Return type:

ChatMessage

async list_messages(chat_id, *, before=None, limit=50)[source]#

See list_messages().

Parameters:

  • chat_id (str)

  • before (str | None)

  • limit (int)

Return type:

Page[ChatMessage]

async update_last_assistant_message(chat_id, *, content, sources=None, finish_reason, error_code=None, error_message=None)[source]#

See update_last_assistant_message().

Parameters:

  • chat_id (str)

  • content (str | None)

  • sources (list[dict[str, Any]] | None)

  • finish_reason (Literal['stop', 'length', 'error', 'interrupted'])

  • error_code (Literal['MODEL_ERROR', 'TIMEOUT', 'RATE_LIMIT', 'CONTEXT_TOO_LONG', 'NETWORK_ERROR', 'SERVICE_UNAVAILABLE', 'INVALID_REQUEST', 'INVALID_RESPONSE', 'INTERNAL_ERROR'] | None)

  • error_message (str | None)

Return type:

ChatMessage | None

class fastrag.query.chat.Session(id, created_at, expires_at, ip_address=None, country=None)[source]#

Bases: object

One anonymous browser identity, owning its chats.

Parameters:
id#

ULID, minted on creation. The bearer credential itself — a caller who holds this id can act as this session — so it is never logged.

Type:

str

created_at#

Derived from id, so ORDER BY id and ORDER BY created_at always agree.

Type:

datetime.datetime

expires_at#

When this session (and, by cascade, its chats and their messages) stops being usable.

Type:

datetime.datetime

ip_address#

The creating request’s peer address, if known.

Type:

str | None

country#

A two-letter country code, if known.

Type:

str | None

fastrag.query.chat.to_llm_history(messages)[source]#

Turn a page of persisted messages into model-facing chat history.

Reorders messages into chronological (oldest-first) order — the order every ChatModel expects — and drops any turn with content is None (an assistant turn that errored before generating anything), which a chat model has nothing to replay. A list_messages page is newest-first, so this is normally the conversion a caller applies right before building a prompt.

Parameters:

messages (Sequence[ChatMessage])

Return type:

list[Message]

Modules

base

The ChatStore contract: sessions, chats, and messages.

postgres

PostgreSQL-backed ChatStore.

schema

The tables behind PostgresChatStore.