diff --git a/examples/foundational/07b-interruptible-langchain.py b/examples/foundational/07b-interruptible-langchain.py index 40082f994..cb375fd0f 100644 --- a/examples/foundational/07b-interruptible-langchain.py +++ b/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", - "{input}"), + MessagesPlaceholder("chat_history"), + ("human", "{input}"), ]) - chain = prompt | llm - lc = LangchainProcessor(chain) + chain = prompt | ChatOpenAI(model="gpt-4o", temperature=0.7) + 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 diff --git a/src/pipecat/services/langchain.py b/src/pipecat/services/langchain.py index 6675005eb..2042071a7 100644 --- a/src/pipecat/services/langchain.py +++ b/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())