Merge pull request #2715 from pipecat-ai/pk/langchain-processor-support-univeral-context

Add support for universal `LLMContext` to `LangchainProcessor`

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

@@ -23,13 +23,8 @@ from pipecat.frames.frames import LLMMessagesUpdateFrame
 from pipecat.pipeline.pipeline import Pipeline
 from pipecat.pipeline.runner import PipelineRunner
 from pipecat.pipeline.task import PipelineParams, PipelineTask
-from pipecat.processors.aggregators.llm_response import (
+from pipecat.processors.aggregators.llm_context import LLMContext
-    LLMAssistantContextAggregator,
+from pipecat.processors.aggregators.llm_response_universal import LLMContextAggregatorPair
-    LLMUserContextAggregator,
-)
-from pipecat.processors.aggregators.openai_llm_context import (
-    OpenAILLMContext,
-)
 from pipecat.processors.frameworks.langchain import LangchainProcessor
 from pipecat.runner.types import RunnerArguments
 from pipecat.runner.utils import create_transport
@@ -106,19 +101,18 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
     )
     lc = LangchainProcessor(history_chain)
 
-    context = OpenAILLMContext()
+    context = LLMContext()
-    tma_in = LLMUserContextAggregator(context=context)
+    context_aggregator = LLMContextAggregatorPair(context)
-    tma_out = LLMAssistantContextAggregator(context=context)
 
     pipeline = Pipeline(
         [
             transport.input(),  # Transport user input
             stt,
-            tma_in,  # User responses
+            context_aggregator.user(),  # User responses
             lc,  # Langchain
             tts,  # TTS
             transport.output(),  # Transport bot output
-            tma_out,  # Assistant spoken responses
+            context_aggregator.assistant(),  # Assistant spoken responses
         ]
     )
 

src/pipecat/processors/frameworks/langchain.py

@@ -12,6 +12,7 @@ from loguru import logger
 
 from pipecat.frames.frames import (
     Frame,
+    LLMContextFrame,
     LLMFullResponseEndFrame,
     LLMFullResponseStartFrame,
     TextFrame,
@@ -64,11 +65,16 @@ class LangchainProcessor(FrameProcessor):
         """
         await super().process_frame(frame, direction)
 
-        if isinstance(frame, OpenAILLMContextFrame):
+        if isinstance(frame, (LLMContextFrame, OpenAILLMContextFrame)):
             # Messages are accumulated on the context as a list of messages.
             # The last one by the human is the one we want to send to the LLM.
             logger.debug(f"Got transcription frame {frame}")
-            text: str = frame.context.messages[-1]["content"]
+            messages = (
+                frame.context.messages
+                if isinstance(frame, OpenAILLMContextFrame)
+                else frame.context.get_messages()
+            )
+            text: str = messages[-1]["content"]
 
             await self._ainvoke(text.strip())
         else: