wip: Example using LC message history

examples/foundational/07b-interruptible-langchain.py

@@ -4,6 +4,7 @@
 # SPDX-License-Identifier: BSD 2-Clause License
 #
 
+
 import asyncio
 import os
 import sys
@@ -27,8 +28,13 @@ from pipecat.vad.silero import SileroVADAnalyzer
 load_dotenv(override=True)
 
 try:
-    from langchain.prompts import ChatPromptTemplate
+    from langchain.prompts import ChatPromptTemplate, MessagesPlaceholder
+    from langchain_community.chat_message_histories import ChatMessageHistory
+    from langchain_core.chat_history import BaseChatMessageHistory
+    from langchain_core.runnables.history import (BaseChatMessageHistory,
+                                                  RunnableWithMessageHistory)
     from langchain_openai import ChatOpenAI
+
 except ModuleNotFoundError as e:
     logger.exception(
         "You need to `pip install langchain_openai` for this example. Also, be sure to set `OPENAI_API_KEY` in the environment variable."
@@ -38,6 +44,14 @@ except ModuleNotFoundError as e:
 logger.remove(0)
 logger.add(sys.stderr, level="DEBUG")
 
+message_store = {}
+
+
+def get_session_history(session_id: str) -> BaseChatMessageHistory:
+    if session_id not in message_store:
+        message_store[session_id] = ChatMessageHistory()
+    return message_store[session_id]
+
 
 async def main(room_url: str, token):
     async with aiohttp.ClientSession() as session:
@@ -59,17 +73,22 @@ async def main(room_url: str, token):
             voice_id=os.getenv("ELEVENLABS_VOICE_ID"),
         )
 
-        llm = ChatOpenAI(model="gpt-4o", temperature=0.7)
         prompt = ChatPromptTemplate.from_messages(
             [
                 ("system",
-                 "Be nice and helpful. Answer very briefly and without special characters like `#` or `*`. Your response will be synthesized to voice and those characters will create unnatural sounds.",
+                 "Be nice and helpful. Answer very briefly and without special characters like `#` or `*`. "
+                 "Your response will be synthesized to voice and those characters will create unnatural sounds.",
                  ),
-                ("human",
+                MessagesPlaceholder("chat_history"),
-                 "{input}"),
+                ("human", "{input}"),
             ])
-        chain = prompt | llm
+        chain = prompt | ChatOpenAI(model="gpt-4o", temperature=0.7)
-        lc = LangchainProcessor(chain)
+        history_chain = RunnableWithMessageHistory(
+            chain,
+            get_session_history,
+            history_messages_key="chat_history",
+            input_messages_key="input")
+        lc = LangchainProcessor(history_chain)
 
         tma_in = LLMUserResponseAggregator()
         tma_out = LLMAssistantResponseAggregator()
@@ -90,6 +109,7 @@ async def main(room_url: str, token):
         @transport.event_handler("on_first_participant_joined")
         async def on_first_participant_joined(transport, participant):
             transport.capture_participant_transcription(participant["id"])
+            lc.set_participant_id(participant["id"])
             # Kick off the conversation.
             # the `LLMMessagesFrame` will be picked up by the LangchainProcessor using
             # only the content of the last message to inject it in the prompt defined

src/pipecat/services/langchain.py

@@ -17,6 +17,10 @@ class LangchainProcessor(FrameProcessor):
         super().__init__()
         self._chain = chain
         self._transcript_key = transcript_key
+        self._participant_id: str | None = None
+
+    def set_participant_id(self, participant_id: str):
+        self._participant_id = participant_id
 
     async def process_frame(self, frame: Frame, direction: FrameDirection):
         if isinstance(frame, LLMMessagesFrame):
@@ -30,7 +34,10 @@ class LangchainProcessor(FrameProcessor):
             await self.push_frame(frame)
 
     async def _invoke(self, text: str):
-        response = await self._chain.ainvoke({self._transcript_key: text})
+        response = await self._chain.ainvoke(
+            {self._transcript_key: text},
+            config={"configurable": {"session_id": self._participant_id}},
+        )
         await self.push_frame(LLMFullResponseStartFrame())
         await self.push_frame(TextFrame(response))
         await self.push_frame(LLMFullResponseEndFrame())
@@ -49,7 +56,10 @@ class LangchainProcessor(FrameProcessor):
         logger.debug(f"Invoking chain with {text}")
         await self.push_frame(LLMFullResponseStartFrame())
         try:
-            async for token in self._chain.astream({self._transcript_key: text}):
+            async for token in self._chain.astream(
+                {self._transcript_key: text},
+                config={"configurable": {"session_id": self._participant_id}},
+            ):
                 await self.push_frame(LLMResponseStartFrame())
                 await self.push_frame(TextFrame(self.__get_token_value(token)))
                 await self.push_frame(LLMResponseEndFrame())