3fi7dZddlmZddlZddlZddlZddlZddlmZm Z m Z ddl Z ddl m Z ddlmZmZmZddl mZej&eZddZdd Zdd Zdd Zdd ZGd de Zy)zClient for persisting chat message history in a Postgres database. This client provides support for both sync and async via psycopg 3. ) annotationsN)ListOptionalSequence)BaseChatMessageHistory) BaseMessagemessage_to_dictmessages_from_dict)sqlcd|d}tjdjtj|tjdjtj|tj|g}|S)z#Make a SQL query to create a table.idx_ _session_ida  CREATE TABLE IF NOT EXISTS {table_name} ( id SERIAL PRIMARY KEY, session_id UUID NOT NULL, message JSONB NOT NULL, created_at TIMESTAMPTZ NOT NULL DEFAULT NOW() );  table_namez_ CREATE INDEX IF NOT EXISTS {index_name} ON {table_name} (session_id); )r index_namer SQLformat Identifier)rr statementss g/var/www/auto_recruiter/arenv/lib/python3.12/site-packages/langchain_postgres/chat_message_histories.py_create_table_and_indexrs |;/J    &CNN:6& 7    &~~j1cnnZ>X  J& crtjdjtj|S)z5Make a SQL query to get messages for a given session.zOSELECT message FROM {table_name} WHERE session_id = %(session_id)s ORDER BY id;rrrs r_get_messages_queryr/s/ 77    fz2f3 4rcrtjdjtj|S)z8Make a SQL query to delete messages for a given session.z;DELETE FROM {table_name} WHERE session_id = %(session_id)s;rrrs r_delete_by_session_id_queryr9s+ 77E  fz2f34rcrtjdjtj|S)z#Make a SQL query to delete a table.z"DROP TABLE IF EXISTS {table_name};rrrs r_delete_table_queryr@s1 777 8 ? ?>>*- @ rcrtjdjtj|S)z%Make a SQL query to insert a message.z>INSERT INTO {table_name} (session_id, message) VALUES (%s, %s)rrrs r_insert_message_queryr!Gs+ 77H  fz2f34rceZdZddd ddZe ddZe ddZeddZe ddZddZ dd Z dd Z dd Z e dd Zejdd ZddZddZy)PostgresChatMessageHistoryN)sync_connectionasync_connectionc|s |s td||_||_ tj|||_t jd|s td||_y#t$rtd|wxYw)arClient for persisting chat message history in a Postgres database, This client provides support for both sync and async via psycopg >=3. The client can create schema in the database and provides methods to add messages, get messages, and clear the chat message history. The schema has the following columns: - id: A serial primary key. - session_id: The session ID for the chat message history. - message: The JSONB message content. - created_at: The timestamp of when the message was created. Messages are retrieved for a given session_id and are sorted by the id (which should be increasing monotonically), and correspond to the order in which the messages were added to the history. The "created_at" column is not returned by the interface, but has been added for the schema so the information is available in the database. A session_id can be used to separate different chat histories in the same table, the session_id should be provided when initializing the client. This chat history client takes in a psycopg connection object (either Connection or AsyncConnection) and uses it to interact with the database. This design allows to reuse the underlying connection object across multiple instantiations of this class, making instantiation fast. This chat history client is designed for prototyping applications that involve chat and are based on Postgres. As your application grows, you will likely need to extend the schema to handle more complex queries. For example, a chat application may involve multiple tables like a user table, a table for storing chat sessions / conversations, and this table for storing chat messages for a given session. The application will require access to additional endpoints like deleting messages by user id, listing conversations by user id or ordering them based on last message time, etc. Feel free to adapt this implementation to suit your application's needs. Args: session_id: The session ID to use for the chat message history table_name: The name of the database table to use sync_connection: An existing psycopg connection instance async_connection: An existing psycopg async connection instance Usage: - Use the create_tables or acreate_tables method to set up the table schema in the database. - Initialize the class with the appropriate session ID, table name, and database connection. - Add messages to the database using add_messages or aadd_messages. - Retrieve messages with get_messages or aget_messages. - Clear the session history with clear or aclear when needed. Note: - At least one of sync_connection or async_connection must be provided. Examples: .. code-block:: python import uuid from langchain_core.messages import SystemMessage, AIMessage, HumanMessage from langchain_postgres import PostgresChatMessageHistory import psycopg # Establish a synchronous connection to the database # (or use psycopg.AsyncConnection for async) sync_connection = psycopg2.connect(conn_info) # Create the table schema (only needs to be done once) table_name = "chat_history" PostgresChatMessageHistory.create_tables(sync_connection, table_name) session_id = str(uuid.uuid4()) # Initialize the chat history manager chat_history = PostgresChatMessageHistory( table_name, session_id, sync_connection=sync_connection ) # Add messages to the chat history chat_history.add_messages([ SystemMessage(content="Meow"), AIMessage(content="woof"), HumanMessage(content="bark"), ]) print(chat_history.messages) z0Must provide sync_connection or async_connectionz9Invalid session id. Session id must be a valid UUID. Got z^\w+$zYInvalid table name. Table name must contain only alphanumeric characters and underscores.N) ValueError _connection _aconnectionuuidUUIDrrematch _table_name)selfr session_idr$r%s r__init__z#PostgresChatMessageHistory.__init__OsT'7OP P*,  IIj ! &xx*-. & KJ<X  s A$$A<ct|}tjd||j5}|D]}|j | ddd|j y#1swYxYwzDCreate the table schema in the database and create relevant indexes.zCreating schema for table %sNrloggerinfocursorexecutecommit) connectionrqueriesr7querys r create_tablesz(PostgresChatMessageHistory.create_tablessi**5 2J?     &F  &u% & &  & &s A%%A.cJKt|}tjd||j4d{}|D]}|j |d{dddd{|j d{y7M717!#1d{7swY1xYw7 wr3r4)r:rr;curr<s racreate_tablesz)PostgresChatMessageHistory.acreate_tabless **5 2J?$$& ) )#  )kk%((( ) ) )!!! )( ) ) ) ) "sb6B#BB#B BB  B#(B )B#B!B#B  B# BB BB#ct|}tjd||j5}|j |ddd|j y#1swYxYw)aLDelete the table schema in the database. WARNING: This will delete the given table from the database including all the database in the table and the schema of the table. Args: connection: The database connection. table_name: The name of the table to create. Dropping table %sNrr5r6r7r8r9)r:rr<r7s r drop_tablez%PostgresChatMessageHistory.drop_tablesX$J/ '4     "F NN5 ! " " "s AA&c<Kt|}tjd||j4d{}|j |d{dddd{|j d{y7F7/7!#1d{7swY1xYw7 w)aNDelete the table schema in the database. WARNING: This will delete the given table from the database including all the database in the table and the schema of the table. Args: connection: Async database connection. table_name: The name of the table to create. rBNrC)r:rr<acurs r adrop_tablez&PostgresChatMessageHistory.adrop_tables$J/ '4$$& & &$,,u% % % & &!!! & % & & & &!sb6BA?BBBB B!B"B9B:BBBB B BBc |j td|Dcgc],}|jtjt |f.}}t |j}|jj5}|j||ddd|jjycc}w#1swY)xYw))Add messages to the chat message history.NzpPlease initialize the PostgresChatMessageHistory with a sync connection or use the aadd_messages method instead.) r(r'rjsondumpsr r!r.r7 executemanyr9r/messagesmessagevaluesr<r7s r add_messagesz'PostgresChatMessageHistory.add_messagess    #R $   tzz/'*BC D  &d&6&67    $ $ & .&   uf - . !  . .s1B4>B99Cc K|j td|Dcgc],}|jtjt |f.}}t |j}|jj4d{}|j||d{dddd{|jjd{ycc}w7V7>70#1d{7swY@xYw7%w)rINzvPlease initialize the PostgresChatMessageHistory with an async connection or use the sync add_messages method instead.) r)r'rrJrKr r!r.r7rLr9rMs r aadd_messagesz(PostgresChatMessageHistory.aadd_messages%s    $X $   tzz/'*BC D  &d&6&67$$++- 4 4$$UF3 3 3 4 4&&(((  4 3 4 4 4 4(spC81C5C8CC8C!CC!# C8.C/!C8C6 C8C!C8!C3'C* (C3/C8cX|j tdt|j}|jj 5}|j |d|j i|jDcgc]}|d }}dddt}|Scc}w#1swYxYw)0Retrieve messages from the chat message history.NzvPlease initialize the PostgresChatMessageHistory with a sync connection or use the async aget_messages method instead.r0r) r(r'rr.r7r8rfetchallr r/r<r7recorditemsrNs r get_messagesz'PostgresChatMessageHistory.get_messages7s    #X  $D$4$45    $ $ & @& NN5<1A1A"B C-3__->?6VAY?E? @&e,@ @ @s1B 8 BB B  B)cK|j tdt|j}|jj 4d{}|j |d|j id{|jd{Dcgc]}|d }}dddd{t}|S7o7K75cc}w7#1d{7swY,xYww)rUNzvPlease initialize the PostgresChatMessageHistory with an async connection or use the sync get_messages method instead.r0r) r)r'rr.r7r8rrVr rWs r aget_messagesz(PostgresChatMessageHistory.aget_messagesHs    $X  $D$4$45$$++- F F..t7G7G(HI I I39??3D-DE6VAYEEE F F&e,  F I-DE F F F Fs~A C B=C"C 3B?4C  C C  CC  C+C,C?C C C C CC CCc"|jS)z$The abstraction required a property.)rZ)r/s rrNz#PostgresChatMessageHistory.messagesXs  ""rcF|j|j|y)z9Clear the stored messages and appends a list of messages.N)clearrQ)r/values rrNz#PostgresChatMessageHistory.messages]s  % rc*|j tdt|j}|jj 5}|j |d|j iddd|jjy#1swY$xYw)5Clear the chat message history for the GIVEN session.NznPlease initialize the PostgresChatMessageHistory with a sync connection or use the async clear method instead.r0)r(r'rr.r7r8rr9r/r<r7s rr_z PostgresChatMessageHistory.clearcs    #P  ,D,<,<=    $ $ & D& NN5<1A1A"B C D ! D Ds B  BcK|j tdt|j}|jj 4d{}|j |d|j id{dddd{|jjd{y7]797+#1d{7swY;xYw7 w)rbNzoPlease initialize the PostgresChatMessageHistory with an async connection or use the sync clear method instead.r0)r)r'rr.r7r8rr9rcs raclearz!PostgresChatMessageHistory.aclearps    $Q  ,D,<,<=$$++- J J..t7G7G(HI I I J J&&((( J I J J J J(sfA C B+C"B13B-4B18 CB/!C%C&C-B1/C1C7B: 8C?C) rstrr0rfr$zOptional[psycopg.Connection]r%z!Optional[psycopg.AsyncConnection]returnNone)r:zpsycopg.Connectionrrfrgrh)r:zpsycopg.AsyncConnectionrrfrgrh)rNzSequence[BaseMessage]rgrh)rgzList[BaseMessage])r`zlist[BaseMessage]rgrh)rgrh)__name__ __module__ __qualname__r1 staticmethodr=r@rDrGrQrSrZr\propertyrNsetterr_rerrr#r#Ns*9=>B&&& 6 &<& &B &      "+ "9< "  " "$"+"9<" ""("&)$" ##__!! " )rr#)rrfrgzList[sql.Composed])rrfrgz sql.Composed)__doc__ __future__rrJloggingr,r*typingrrrpsycopglangchain_core.chat_historyrlangchain_core.messagesrr r r getLoggerrir5rrrrr!r#rorrrxsj #  ++>TT   8 $2444m)!7m)r