Merge pull request #199 from pipecat-ai/aleix/langchain-changelog

move LangchainProcessor to processors/frameworks and update CHANGELOG

CHANGELOG.md

@@ -19,6 +19,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   the weather in Los Angeles, the LLM will call your function.
   See https://platform.openai.com/docs/guides/function-calling
 
+- Added new `LangchainProcessor`.
+
 - Added Cartesia TTS support (https://cartesia.ai/)
 
 ### Fixed

src/pipecat/processors/frameworks/langchain.py

@@ -1,14 +1,23 @@
-import sys
+#
+# Copyright (c) 2024, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
 from typing import Union
 
-from loguru import logger
+from pipecat.frames.frames import (
-
+    Frame,
-from pipecat.frames.frames import (Frame, LLMFullResponseEndFrame,
+    LLMFullResponseEndFrame,
-                                   LLMFullResponseStartFrame, LLMMessagesFrame,
+    LLMFullResponseStartFrame,
-                                   LLMResponseEndFrame, LLMResponseStartFrame,
+    LLMMessagesFrame,
-                                   TextFrame)
+    LLMResponseEndFrame,
+    LLMResponseStartFrame,
+    TextFrame)
 from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
 
+from loguru import logger
+
 try:
     from langchain_core.messages import AIMessageChunk
     from langchain_core.runnables import Runnable
@@ -38,26 +47,17 @@ class LangchainProcessor(FrameProcessor):
 
             await self._ainvoke(text.strip())
         else:
-            await self.push_frame(frame)
+            await self.push_frame(frame, direction)
-
-    async def _invoke(self, text: str):
-        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())
 
     @staticmethod
-    def __get_token_value(text: Union[str, AIMessageChunk]) -> str | None:
+    def __get_token_value(text: Union[str, AIMessageChunk]) -> str:
         match text:
             case str():
                 return text
             case AIMessageChunk():
                 return text.content
             case _:
-                return None
+                return ""
 
     async def _ainvoke(self, text: str):
         logger.debug(f"Invoking chain with {text}")
@@ -72,8 +72,6 @@ class LangchainProcessor(FrameProcessor):
                 await self.push_frame(LLMResponseEndFrame())
         except GeneratorExit:
             logger.warning("Generator was closed prematurely")
-            raise  # Re-raise to ensure proper generator closure
         except Exception as e:
             logger.error(f"An unknown error occurred: {e}")
-            raise
         await self.push_frame(LLMFullResponseEndFrame())

tests/test_langchain.py

@@ -1,7 +1,10 @@
-import unittest
+#
+# Copyright (c) 2024, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
 
-from langchain.prompts import ChatPromptTemplate
+import unittest
-from langchain_core.language_models import FakeStreamingListLLM
 
 from pipecat.frames.frames import (LLMFullResponseEndFrame,
                                    LLMFullResponseStartFrame, StopTaskFrame,
@@ -14,7 +17,10 @@ from pipecat.pipeline.task import PipelineTask
 from pipecat.processors.aggregators.llm_response import (
     LLMAssistantResponseAggregator, LLMUserResponseAggregator)
 from pipecat.processors.frame_processor import FrameProcessor
-from pipecat.services.langchain import LangchainProcessor
+from pipecat.processors.frameworks.langchain import LangchainProcessor
+
+from langchain.prompts import ChatPromptTemplate
+from langchain_core.language_models import FakeStreamingListLLM
 
 
 class TestLangchain(unittest.IsolatedAsyncioTestCase):