diff --git a/libs/community/langchain_community/tools/__init__.py b/libs/community/langchain_community/tools/__init__.py index de486cfb..9c0ec2c5 100644 --- a/libs/community/langchain_community/tools/__init__.py +++ b/libs/community/langchain_community/tools/__init__.py @@ -317,6 +317,9 @@ TavilyAnswer, TavilySearchResults, ) + from langchain_community.tools.typecast.text2speech import ( + TypecastText2SpeechTool, + ) from langchain_community.tools.vectorstore.tool import ( VectorStoreQATool, VectorStoreQAWithSourcesTool, @@ -402,6 +405,7 @@ "EdenAiTextToSpeechTool", "EdenaiTool", "ElevenLabsText2SpeechTool", + "TypecastText2SpeechTool", "ExtractHyperlinksTool", "ExtractTextTool", "FileSearchTool", @@ -556,6 +560,7 @@ "EdenAiTextToSpeechTool": "langchain_community.tools.edenai", "EdenaiTool": "langchain_community.tools.edenai", "ElevenLabsText2SpeechTool": "langchain_community.tools.eleven_labs.text2speech", + "TypecastText2SpeechTool": "langchain_community.tools.typecast.text2speech", "ExtractHyperlinksTool": "langchain_community.tools.playwright", "ExtractTextTool": "langchain_community.tools.playwright", "FileSearchTool": "langchain_community.tools.file_management", diff --git a/libs/community/langchain_community/tools/typecast/__init__.py b/libs/community/langchain_community/tools/typecast/__init__.py new file mode 100644 index 00000000..86708d74 --- /dev/null +++ b/libs/community/langchain_community/tools/typecast/__init__.py @@ -0,0 +1,5 @@ +"""Typecast Services Tools.""" + +from langchain_community.tools.typecast.text2speech import TypecastText2SpeechTool + +__all__ = ["TypecastText2SpeechTool"] diff --git a/libs/community/langchain_community/tools/typecast/text2speech.py b/libs/community/langchain_community/tools/typecast/text2speech.py new file mode 100644 index 00000000..b16abd5d --- /dev/null +++ b/libs/community/langchain_community/tools/typecast/text2speech.py @@ -0,0 +1,118 @@ +import tempfile +import logging +from typing import Any, Dict, Optional + +from langchain_core.callbacks import CallbackManagerForToolRun +from langchain_core.tools import BaseTool +from langchain_core.utils import get_from_dict_or_env +from pydantic import model_validator + +logger = logging.getLogger(__name__) + + +def _import_typecast() -> Any: + try: + import typecast + except ImportError as e: + raise ImportError( + "Cannot import typecast, please install `pip install typecast-python`." + ) from e + return typecast + + +class TypecastText2SpeechTool(BaseTool): + """Tool that queries the Typecast Text2Speech API. + + In order to set this up, follow instructions at: + https://typecast.ai/docs/overview + """ + + model: str = "ssfm-v21" + voice_id: str = "tc_62a8975e695ad26f7fb514d1" + language: Optional[str] = None + emotion_preset: str = "normal" + emotion_intensity: float = 1.0 + audio_format: str = "wav" + + name: str = "typecast_text2speech" + description: str = ( + "A wrapper around Typecast Text2Speech. " + "Useful for when you need to convert text to speech. " + "It supports 27+ languages including English, Korean, Spanish, Japanese, " + "Chinese, and many more with emotion control capabilities. " + ) + + @model_validator(mode="before") + @classmethod + def validate_environment(cls, values: Dict) -> Any: + """Validate that api key exists in environment.""" + _ = get_from_dict_or_env(values, "typecast_api_key", "TYPECAST_API_KEY") + + return values + + def _run( + self, query: str, run_manager: Optional[CallbackManagerForToolRun] = None + ) -> str: + """Use the tool.""" + typecast = _import_typecast() + + try: + # Get API key from environment + import os + + api_key = os.environ.get("TYPECAST_API_KEY") + + # Initialize client + client = typecast.Typecast(api_key=api_key) + + # Create TTS request + request = typecast.TTSRequest( + text=query, + model=self.model, + voice_id=self.voice_id, + language=self.language, + prompt=typecast.Prompt( + emotion_preset=self.emotion_preset, + emotion_intensity=self.emotion_intensity, + ), + output=typecast.Output(audio_format=self.audio_format), + ) + + # Convert text to speech + response = client.text_to_speech(request) + + # Save to temporary file + suffix = f".{self.audio_format}" + with tempfile.NamedTemporaryFile( + mode="wb", suffix=suffix, delete=False + ) as f: + f.write(response.audio_data) + + return f.name + except Exception as e: + raise RuntimeError(f"Error while running TypecastText2SpeechTool: {e}") + + def play(self, speech_file: str) -> None: + """Play the text as speech.""" + try: + import sounddevice as sd + import soundfile as sf + except ImportError as e: + logger.error( + "Cannot import sounddevice or soundfile. " + "Please install them with `pip install sounddevice soundfile`. " + f"Error: {e}" + ) + return + + try: + data, samplerate = sf.read(speech_file) + + # Get the current output device + output_device = sd.default.device[1] # [input, output] + + # Play on the current output device + sd.play(data, samplerate, device=output_device) + sd.wait() + except Exception as e: + logger.error(f"Error while playing audio: {e}") diff --git a/libs/community/tests/integration_tests/tools/test_typecast.py b/libs/community/tests/integration_tests/tools/test_typecast.py new file mode 100644 index 00000000..349aff03 --- /dev/null +++ b/libs/community/tests/integration_tests/tools/test_typecast.py @@ -0,0 +1,172 @@ +"""Integration tests for Typecast Text2Speech Tool.""" + +import os + +import pytest + +from langchain_community.tools.typecast import TypecastText2SpeechTool + + +@pytest.mark.requires("typecast") +class TestTypecastText2SpeechTool: + """Integration tests for TypecastText2SpeechTool.""" + + def test_typecast_text_to_speech_basic(self) -> None: + """Test basic text to speech conversion.""" + # Skip if no API key is set + if not os.environ.get("TYPECAST_API_KEY"): + pytest.skip("TYPECAST_API_KEY not set") + + tool = TypecastText2SpeechTool() + text = "Hello world! This is a test." + + # Convert text to speech + audio_file = tool.run(text) + + # Verify file was created + assert audio_file is not None + assert isinstance(audio_file, str) + assert os.path.exists(audio_file) + assert audio_file.endswith(".wav") + + # Clean up + if os.path.exists(audio_file): + os.remove(audio_file) + + def test_typecast_text_to_speech_with_emotion(self) -> None: + """Test text to speech with emotion settings.""" + # Skip if no API key is set + if not os.environ.get("TYPECAST_API_KEY"): + pytest.skip("TYPECAST_API_KEY not set") + + tool = TypecastText2SpeechTool( + emotion_preset="happy", + emotion_intensity=1.5, + ) + text = "I am so excited to show you this!" + + # Convert text to speech + audio_file = tool.run(text) + + # Verify file was created + assert audio_file is not None + assert isinstance(audio_file, str) + assert os.path.exists(audio_file) + + # Clean up + if os.path.exists(audio_file): + os.remove(audio_file) + + def test_typecast_text_to_speech_mp3_format(self) -> None: + """Test text to speech with MP3 format.""" + # Skip if no API key is set + if not os.environ.get("TYPECAST_API_KEY"): + pytest.skip("TYPECAST_API_KEY not set") + + tool = TypecastText2SpeechTool(audio_format="mp3") + text = "This should be in MP3 format." + + # Convert text to speech + audio_file = tool.run(text) + + # Verify file was created + assert audio_file is not None + assert isinstance(audio_file, str) + assert os.path.exists(audio_file) + assert audio_file.endswith(".mp3") + + # Clean up + if os.path.exists(audio_file): + os.remove(audio_file) + + def test_typecast_text_to_speech_korean(self) -> None: + """Test text to speech with Korean language.""" + # Skip if no API key is set + if not os.environ.get("TYPECAST_API_KEY"): + pytest.skip("TYPECAST_API_KEY not set") + + tool = TypecastText2SpeechTool(language="kor") + text = "안녕하세요. 타입캐스트 테스트입니다." + + # Convert text to speech + audio_file = tool.run(text) + + # Verify file was created + assert audio_file is not None + assert isinstance(audio_file, str) + assert os.path.exists(audio_file) + + # Clean up + if os.path.exists(audio_file): + os.remove(audio_file) + + def test_typecast_with_agent(self) -> None: + """Test TypecastText2SpeechTool with a LangChain agent.""" + # Skip if no API key is set + if not os.environ.get("TYPECAST_API_KEY") or not os.environ.get( + "OPENAI_API_KEY" + ): + pytest.skip("TYPECAST_API_KEY or OPENAI_API_KEY not set") + + try: + from langchain.agents import create_agent + from langchain_openai import ChatOpenAI + except ImportError as e: + pytest.skip(f"Required packages not installed: {e}") + + # Create LLM and tools + llm = ChatOpenAI(model="gpt-3.5-turbo-0125", temperature=0) + tools = [TypecastText2SpeechTool()] + + # Create agent + agent = create_agent( + model=llm, + tools=tools, + system_prompt="You are a helpful assistant", + ) + + # Run the agent + query = ( + "Convert this text to speech: 'Hello from LangChain agent!' " + "and tell me the file path." + ) + + # Invoke the agent + # The input schema for create_agent graph expects 'messages' + response = agent.invoke({"messages": [{"role": "user", "content": query}]}) + + # Verify response + # The output state contains 'messages'. The last message should be AIMessage. + messages = response["messages"] + last_message = messages[-1] + output = last_message.content + print(f"\nAgent response: {output}") + + # The output should contain the file path or mention it + assert ".wav" in output or ".mp3" in output + + # Try to clean up any audio files created during the test + import tempfile + import time + + temp_dir = tempfile.gettempdir() + current_time = time.time() + + # Find the file path in the output if possible + import re + match = re.search(r'/tmp/[\w-]+\.(wav|mp3)', output) + if match: + file_path = match.group(0) + if os.path.exists(file_path): + os.remove(file_path) + else: + # Fallback cleanup + for file in os.listdir(temp_dir): + if file.endswith((".wav", ".mp3")): + file_path = os.path.join(temp_dir, file) + try: + # Only remove files created in the last 10 seconds + if os.path.getmtime(file_path) > (current_time - 10): + os.remove(file_path) + except Exception: + pass diff --git a/libs/community/tests/unit_tests/tools/test_imports.py b/libs/community/tests/unit_tests/tools/test_imports.py index e93d224c..bde16a3f 100644 --- a/libs/community/tests/unit_tests/tools/test_imports.py +++ b/libs/community/tests/unit_tests/tools/test_imports.py @@ -133,6 +133,7 @@ "TavilyAnswer", "TavilySearchResults", "Tool", + "TypecastText2SpeechTool", "VectorStoreQATool", "VectorStoreQAWithSourcesTool", "WikipediaQueryRun", diff --git a/libs/community/tests/unit_tests/tools/test_typecast.py b/libs/community/tests/unit_tests/tools/test_typecast.py new file mode 100644 index 00000000..2a4dc0a4 --- /dev/null +++ b/libs/community/tests/unit_tests/tools/test_typecast.py @@ -0,0 +1,130 @@ +"""Unit tests for Typecast Text2Speech Tool.""" + +from unittest.mock import MagicMock, patch + +import pytest +from langchain_core.tools import BaseTool + +from langchain_community.tools.typecast import TypecastText2SpeechTool + + +@patch.dict("os.environ", {"TYPECAST_API_KEY": "test_api_key"}) +def test_typecast_tool_initialization() -> None: + """Test TypecastText2SpeechTool initialization.""" + tool = TypecastText2SpeechTool() + assert isinstance(tool, BaseTool) + assert tool.name == "typecast_text2speech" + assert tool.model == "ssfm-v21" + assert tool.voice_id == "tc_62a8975e695ad26f7fb514d1" + assert tool.emotion_preset == "normal" + assert tool.emotion_intensity == 1.0 + assert tool.audio_format == "wav" + + +@patch.dict("os.environ", {"TYPECAST_API_KEY": "test_api_key"}) +def test_typecast_tool_custom_parameters() -> None: + """Test TypecastText2SpeechTool with custom parameters.""" + tool = TypecastText2SpeechTool( + model="ssfm-v20", + voice_id="custom_voice_id", + language="kor", + emotion_preset="happy", + emotion_intensity=1.5, + audio_format="mp3", + ) + assert tool.model == "ssfm-v20" + assert tool.voice_id == "custom_voice_id" + assert tool.language == "kor" + assert tool.emotion_preset == "happy" + assert tool.emotion_intensity == 1.5 + assert tool.audio_format == "mp3" + + +@patch.dict("os.environ", {"TYPECAST_API_KEY": "test_api_key"}) +def test_typecast_tool_validation() -> None: + """Test that tool validates API key presence.""" + tool = TypecastText2SpeechTool() + assert tool is not None + + +def test_typecast_tool_validation_missing_key() -> None: + """Test that tool raises error when API key is missing.""" + with patch.dict("os.environ", {}, clear=True): + with pytest.raises(ValueError): + TypecastText2SpeechTool() + + +@patch.dict("os.environ", {"TYPECAST_API_KEY": "test_api_key"}) +@patch("langchain_community.tools.typecast.text2speech._import_typecast") +def test_typecast_tool_run(mock_import: MagicMock) -> None: + """Test TypecastText2SpeechTool._run method.""" + # Mock the typecast module + mock_typecast = MagicMock() + mock_import.return_value = mock_typecast + + # Mock the client and response + mock_client = MagicMock() + mock_response = MagicMock() + mock_response.audio_data = b"fake_audio_data" + + mock_typecast.Typecast.return_value = mock_client + mock_client.text_to_speech.return_value = mock_response + + # Mock TTSRequest, Prompt, and Output + mock_typecast.TTSRequest = MagicMock() + mock_typecast.Prompt = MagicMock() + mock_typecast.Output = MagicMock() + + # Run the tool + tool = TypecastText2SpeechTool() + result = tool._run("Hello world") + + # Verify the result is a file path + assert isinstance(result, str) + assert result.endswith(".wav") + + # Verify the client was called correctly + mock_typecast.Typecast.assert_called_once_with(api_key="test_api_key") + mock_client.text_to_speech.assert_called_once() + + +@patch.dict("os.environ", {"TYPECAST_API_KEY": "test_api_key"}) +@patch("langchain_community.tools.typecast.text2speech._import_typecast") +def test_typecast_tool_run_with_error(mock_import: MagicMock) -> None: + """Test TypecastText2SpeechTool._run method with error.""" + # Mock the typecast module to raise an exception + mock_typecast = MagicMock() + mock_import.return_value = mock_typecast + mock_typecast.Typecast.side_effect = Exception("API Error") + + # Run the tool and expect a RuntimeError + tool = TypecastText2SpeechTool() + with pytest.raises( + RuntimeError, match="Error while running TypecastText2SpeechTool" + ): + tool._run("Hello world") + + +def test_import_typecast_missing() -> None: + """Test that _import_typecast raises ImportError when typecast is not installed.""" + with patch.dict("sys.modules", {"typecast": None}): + # Remove typecast from sys.modules temporarily + import sys + + import langchain_community.tools.typecast.text2speech as tts_module + + typecast_backup = sys.modules.pop("typecast", None) + try: + with patch.object(tts_module, "_import_typecast") as mock_import: + mock_import.side_effect = ImportError( + "Cannot import typecast, please install " + "`pip install typecast-python`." + ) + with pytest.raises( + ImportError, match="Cannot import typecast, please install" + ): + mock_import() + finally: + # Restore typecast if it was present + if typecast_backup is not None: + sys.modules["typecast"] = typecast_backup