From f087151db793076c6747f00e64f05b25191ffa6e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Fri, 31 May 2024 14:00:07 -0700 Subject: [PATCH] move LangchainProcessor to processors/frameworks and update CHANGELOG --- CHANGELOG.md | 2 + src/pipecat/processors/filters/__init__.py | 0 src/pipecat/processors/frameworks/__init__.py | 0 .../frameworks}/langchain.py | 40 +++++++++---------- src/pipecat/serializers/__init__.py | 0 src/pipecat/transports/__init__.py | 0 src/pipecat/transports/network/__init__.py | 0 src/pipecat/transports/services/__init__.py | 0 tests/test_langchain.py | 14 +++++-- 9 files changed, 31 insertions(+), 25 deletions(-) create mode 100644 src/pipecat/processors/filters/__init__.py create mode 100644 src/pipecat/processors/frameworks/__init__.py rename src/pipecat/{services => processors/frameworks}/langchain.py (73%) create mode 100644 src/pipecat/serializers/__init__.py create mode 100644 src/pipecat/transports/__init__.py create mode 100644 src/pipecat/transports/network/__init__.py create mode 100644 src/pipecat/transports/services/__init__.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 432eaf9bb..3f1bb96b3 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/src/pipecat/processors/filters/__init__.py b/src/pipecat/processors/filters/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/src/pipecat/processors/frameworks/__init__.py b/src/pipecat/processors/frameworks/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/src/pipecat/services/langchain.py b/src/pipecat/processors/frameworks/langchain.py similarity index 73% rename from src/pipecat/services/langchain.py rename to src/pipecat/processors/frameworks/langchain.py index 83522a16c..bac39cf26 100644 --- a/src/pipecat/services/langchain.py +++ b/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, LLMFullResponseEndFrame, - LLMFullResponseStartFrame, LLMMessagesFrame, - LLMResponseEndFrame, LLMResponseStartFrame, - TextFrame) +from pipecat.frames.frames import ( + Frame, + LLMFullResponseEndFrame, + LLMFullResponseStartFrame, + LLMMessagesFrame, + 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) - - 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()) + await self.push_frame(frame, direction) @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()) diff --git a/src/pipecat/serializers/__init__.py b/src/pipecat/serializers/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/src/pipecat/transports/__init__.py b/src/pipecat/transports/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/src/pipecat/transports/network/__init__.py b/src/pipecat/transports/network/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/src/pipecat/transports/services/__init__.py b/src/pipecat/transports/services/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/tests/test_langchain.py b/tests/test_langchain.py index 0f3ccdc86..b33a38c0c 100644 --- a/tests/test_langchain.py +++ b/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 -from langchain_core.language_models import FakeStreamingListLLM +import unittest 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):