wx44wx/pipecat
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

tests: implement langchain tests with run_test()

Browse Source
This commit is contained in:
Aleix Conchillo Flaqué
2025-02-11 22:14:27 -08:00
parent 1b7dfe8126
commit 8bdd7ed0ed
1 changed files with 21 additions and 27 deletions
Download Patch File Download Diff File Expand all files Collapse all files

48
tests/test_langchain.py
View File

@@ -10,23 +10,22 @@ from langchain.prompts import ChatPromptTemplate
from langchain_core.language_models import FakeStreamingListLLM from langchain_core.language_models import FakeStreamingListLLM
from pipecat.frames.frames import ( from pipecat.frames.frames import (
EndFrame,
LLMFullResponseEndFrame, LLMFullResponseEndFrame,
LLMFullResponseStartFrame, LLMFullResponseStartFrame,
LLMMessagesFrame,
TextFrame, TextFrame,
TranscriptionFrame, TranscriptionFrame,
UserStartedSpeakingFrame, UserStartedSpeakingFrame,
UserStoppedSpeakingFrame, UserStoppedSpeakingFrame,
) )
from pipecat.pipeline.pipeline import Pipeline 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_response import (
LLMAssistantResponseAggregator, LLMAssistantResponseAggregator,
LLMUserResponseAggregator, LLMUserResponseAggregator,
) )
from pipecat.processors.frame_processor import FrameProcessor from pipecat.processors.frame_processor import FrameProcessor
from pipecat.processors.frameworks.langchain import LangchainProcessor from pipecat.processors.frameworks.langchain import LangchainProcessor
from pipecat.tests.utils import SleepFrame, run_test
class TestLangchain(unittest.IsolatedAsyncioTestCase): class TestLangchain(unittest.IsolatedAsyncioTestCase):
@@ -64,31 +63,26 @@ class TestLangchain(unittest.IsolatedAsyncioTestCase):
self.mock_proc = self.MockProcessor("token_collector") self.mock_proc = self.MockProcessor("token_collector")
tma_in = LLMUserResponseAggregator(messages) tma_in = LLMUserResponseAggregator(messages)
tma_out = LLMAssistantResponseAggregator(messages) tma_out = LLMAssistantResponseAggregator(messages, expect_stripped_words=False)
pipeline = Pipeline( pipeline = Pipeline([tma_in, proc, self.mock_proc, tma_out])
[
tma_in, frames_to_send = [
proc, UserStartedSpeakingFrame(),
self.mock_proc, TranscriptionFrame(text="Hi World", user_id="user", timestamp="now"),
tma_out, SleepFrame(),
] UserStoppedSpeakingFrame(),
]
expected_down_frames = [
UserStartedSpeakingFrame,
UserStoppedSpeakingFrame,
LLMMessagesFrame,
]
await run_test(
pipeline,
frames_to_send=frames_to_send,
expected_down_frames=expected_down_frames,
) )
task = PipelineTask(pipeline, PipelineParams(allow_interruptions=False))
await task.queue_frames(
[
UserStartedSpeakingFrame(),
TranscriptionFrame(text="Hi World", user_id="user", timestamp="now"),
UserStoppedSpeakingFrame(),
EndFrame(),
]
)
runner = PipelineRunner()
await runner.run(task)
self.assertEqual("".join(self.mock_proc.token), self.expected_response) self.assertEqual("".join(self.mock_proc.token), self.expected_response)
# TODO: Address this issue self.assertEqual(tma_out.messages[-1]["content"], self.expected_response)
# This next one would fail with:
# AssertionError: ' H e l l o d e a r h u m a n' != 'Hello dear human'
# self.assertEqual(tma_out.messages[-1]["content"], self.expected_response)