LLMと電話を組み合わせたパッケージvocodeの中身を勉強する②
langchainについてのパートをchatGPTに解説させました。
vocode.py
import logging
import os
from langchain.agents import tool
from dotenv import load_dotenv
from vocode.streaming.models.message import BaseMessage
from call_transcript_utils import delete_transcript, get_transcript
load_dotenv()
from vocode.streaming.telephony.conversation.outbound_call import OutboundCall
from vocode.streaming.telephony.config_manager.redis_config_manager import (
RedisConfigManager,
)
from vocode.streaming.models.agent import ChatGPTAgentConfig
import time
@tool("call phone number")
def call_phone_number(input: str) -> str:
"""calls a phone number as a bot and returns a transcript of the conversation.
the input to this tool is a pipe separated list of a phone number, a prompt, and the first thing the bot should say.
The prompt should instruct the bot with what to do on the call and be in the 3rd person,
like 'the assistant is performing this task' instead of 'perform this task'.
for example, `+15555555555|the assistant is explaining the meaning of life|i'm going to tell you the meaning of life` will call +15555555555, say 'i'm going to tell you the meaning of life', and instruct the assistant to tell the human what the meaning of life is.
"""
phone_number, prompt, initial_message = input.split("|", 2)
call = OutboundCall(
base_url=os.getenv("TELEPHONY_SERVER_BASE_URL"),
to_phone=phone_number,
from_phone=os.getenv("OUTBOUND_CALLER_NUMBER"),
config_manager=RedisConfigManager(),
agent_config=ChatGPTAgentConfig(
prompt_preamble=prompt,
end_conversation_on_goodbye=True,
initial_message=BaseMessage(text=initial_message)),
logger=logging.Logger("call_phone_number"),
)
call.start()
while True:
maybe_transcript = get_transcript(call.conversation_id)
if maybe_transcript:
delete_transcript(call.conversation_id)
return maybe_transcript
else:
time.sleep(1)
このコードは、電話番号に電話をかけるとともに、対話の要約を生成して返すPythonプログラムです。具体的には、以下の手順で実行されます。
必要なモジュールやクラスをインポートします。
環境変数を読み込みます。
関数 call_phone_number を定義します。この関数は、パイプで区切られた電話番号、プロンプト、最初のメッセージを受け取ります。
関数 call_phone_number の詳細は以下の通りです。
関数の入力から電話番号、プロンプト、最初のメッセージを取得します。
OutboundCall インスタンスを作成し、必要な情報を渡します。これには、電話番号、発信者番号、設定マネージャー、エージェント設定、およびロガーが含まれます。
電話を開始するために call.start() を呼び出します。
get_transcript を使って会話の要約を取得し、要約が利用可能になるまで待ちます。要約が利用可能になったら、delete_transcript を使って要約を削除し、要約を返します。
このプログラムは、指定された電話番号に電話をかけ、最初のメッセージを伝えた後、与えられたプロンプトに基づいて対話を行います。対話が終了すると、その要約を返します。たとえば、電話番号に +15555555555、プロンプトに the assistant is explaining the meaning of life、最初のメッセージに i'm going to tell you the meaning of life を入力すると、指定された電話番号に電話をかけ、対話が終了したらその要約を返します。
callback_handler.py
import threading
from typing import Any, Dict, List, Optional, Union
from langchain.callbacks.base import BaseCallbackHandler
from langchain.schema import AgentAction, AgentFinish, LLMResult
from vocode.turn_based.output_device.speaker_output import SpeakerOutput
from vocode.turn_based.synthesizer.base_synthesizer import BaseSynthesizer
class VocodeCallbackHandler(BaseCallbackHandler):
"""Custom CallbackHandler."""
def __init__(self, synthesizer: BaseSynthesizer) -> None:
super().__init__()
self.output_device = SpeakerOutput.from_default_device()
self.synthesizer = synthesizer
def _speak_in_thread(self, text: str) -> None:
thread = threading.Thread(target=lambda: self.output_device.send_audio(self.synthesizer.synthesize(text)))
thread.start()
def on_llm_start(
self, serialized: Dict[str, Any], prompts: List[str], **kwargs: Any
) -> None:
pass
def on_llm_end(self, response: LLMResult, **kwargs: Any) -> None:
pass
def on_llm_new_token(self, token: str, **kwargs: Any) -> None:
pass
def on_llm_error(
self, error: Union[Exception, KeyboardInterrupt], **kwargs: Any
) -> None:
pass
def on_chain_start(
self, serialized: Dict[str, Any], inputs: Dict[str, Any], **kwargs: Any
) -> None:
pass
def on_chain_end(self, outputs: Dict[str, Any], **kwargs: Any) -> None:
pass
def on_chain_error(
self, error: Union[Exception, KeyboardInterrupt], **kwargs: Any
) -> None:
pass
def on_tool_start(
self,
serialized: Dict[str, Any],
input_str: str,
**kwargs: Any,
) -> None:
pass
def on_agent_action(
self, action: AgentAction, color: Optional[str] = None, **kwargs: Any
) -> Any:
try:
thought = action.log.split("Action:")[0].replace("Thought: ", "")
self._speak_in_thread(thought)
except:
print("BEGIN LOG\n{}\nEND LOG".format(action.log))
def on_tool_end(
self,
output: str,
color: Optional[str] = None,
observation_prefix: Optional[str] = None,
llm_prefix: Optional[str] = None,
**kwargs: Any,
) -> None:
pass
def on_tool_error(
self, error: Union[Exception, KeyboardInterrupt], **kwargs: Any
) -> None:
pass
def on_text(
self,
text: str,
color: Optional[str] = None,
end: str = "",
**kwargs: Optional[str],
) -> None:
pass
def on_agent_finish(
self, finish: AgentFinish, color: Optional[str] = None, **kwargs: Any
) -> None:
try:
final_answer = finish.log.split("Final Answer:")[1].strip()
self._speak_in_thread(final_answer)
except:
pass
このコードは、VocodeCallbackHandlerというカスタムのコールバックハンドラクラスを定義しています。このクラスは、テキストから音声への変換と、音声の再生を行うことができます。具体的には、以下の機能を提供します。
音声シンセサイザを使用してテキストを音声に変換し、再生する機能。
エージェントのアクションや結果に基づいて音声を再生する機能。
このクラスはBaseCallbackHandlerを継承し、以下のメソッドをオーバーライドまたは実装しています。
init: コンストラクタで、音声シンセサイザを引数として受け取り、SpeakerOutput インスタンスを作成します。
_speak_in_thread: 音声を生成して再生するための内部メソッド。新しいスレッドで音声シンセサイザを使ってテキストを音声に変換し、output_deviceで音声を再生します。
on_llm_start, on_llm_end, on_llm_new_token, on_llm_error, on_chain_start, on_chain_end, on_chain_error, on_tool_start, on_tool_end, on_tool_error, on_text: これらのメソッドは、デフォルトの実装を持つか、何もしないため、特に説明は不要です。
on_agent_action: エージェントのアクションが発生したときに呼び出されるメソッド。アクションのログから思考内容を抽出し、それを音声に変換して再生します。思考内容が見つからない場合は、アクションのログ全体を表示します。
on_agent_finish: エージェントが終了したときに呼び出されるメソッド。終了ログから最終回答を抽出し、それを音声に変換して再生します。最終回答が見つからない場合は、何もしません。
このクラスは、対話エージェントが行動を取るたびに、その行動に基づいて音声を生成して再生することができます。また、対話エージェントが終了した際には、最終的な回答を音声で再生します。
call_transcript_utils.py
import os
from typing import Optional
CALL_TRANSCRIPTS_DIR = os.path.join(os.path.dirname(__file__), "call_transcripts")
def add_transcript(conversation_id: str, transcript: str) -> None:
transcript_path = os.path.join(CALL_TRANSCRIPTS_DIR, "{}.txt".format(conversation_id))
with open(transcript_path, "a") as f:
f.write(transcript)
def get_transcript(conversation_id: str) -> Optional[str]:
transcript_path = os.path.join(CALL_TRANSCRIPTS_DIR, "{}.txt".format(conversation_id))
if os.path.exists(transcript_path):
with open(transcript_path, "r") as f:
return f.read()
def delete_transcript(conversation_id: str) -> bool:
transcript_path = os.path.join(CALL_TRANSCRIPTS_DIR, "{}.txt".format(conversation_id))
if os.path.exists(transcript_path):
os.remove(transcript_path)
return True
return False
このコードは、電話会話の要約を取り扱うための3つの関数を提供しています。CALL_TRANSCRIPTS_DIR というディレクトリに、要約のテキストファイルが保存されます。
add_transcript(conversation_id: str, transcript: str) -> None: この関数は、会話IDと要約を引数として受け取り、要約をテキストファイルに追記します。ファイルのパスは、CALL_TRANSCRIPTS_DIRディレクトリ内にある、{conversation_id}.txt という名前のファイルです。
get_transcript(conversation_id: str) -> Optional[str]: この関数は、会話IDを引数として受け取り、対応する要約が存在すればそれを返します。要約が存在しない場合は、None を返します。要約のテキストファイルは、CALL_TRANSCRIPTS_DIRディレクトリ内にある、{conversation_id}.txt という名前のファイルです。
delete_transcript(conversation_id: str) -> bool: この関数は、会話IDを引数として受け取り、対応する要約のテキストファイルを削除します。要約のファイルが正常に削除された場合は、True を返します。ファイルが存在しない場合は、False を返します。要約のテキストファイルは、CALL_TRANSCRIPTS_DIRディレクトリ内にある、{conversation_id}.txt という名前のファイルです。
これらの関数を使用することで、電話会話の要約を保存、取得、削除することができます。これらの関数は、電話会話の要約を維持し、後で参照できるようにするために使用されることが想定されています。
import logging
import os
import typing
from fastapi import FastAPI
from vocode.streaming.models.events import Event, EventType, TranscriptCompleteEvent
from vocode.streaming.utils import events_manager
from vocode.streaming.telephony.config_manager.redis_config_manager import (
RedisConfigManager,
)
from vocode.streaming.telephony.server.base import TelephonyServer
from call_transcript_utils import add_transcript
app = FastAPI(docs_url=None)
logging.basicConfig()
logger = logging.getLogger(__name__)
logger.setLevel(logging.DEBUG)
class EventsManager(events_manager.EventsManager):
def __init__(self):
super().__init__(subscriptions=[EventType.TRANSCRIPT_COMPLETE])
def handle_event(self, event: Event):
if event.type == EventType.TRANSCRIPT_COMPLETE:
transcript_complete_event = typing.cast(TranscriptCompleteEvent, event)
add_transcript(transcript_complete_event.conversation_id, transcript_complete_event.transcript)
config_manager = RedisConfigManager()
BASE_URL = os.getenv("TELEPHONY_SERVER_BASE_URL")
telephony_server = TelephonyServer(
base_url=BASE_URL,
config_manager=config_manager,
inbound_call_configs=[],
events_manager=EventsManager(),
logger=logger,
)
app.include_router(telephony_server.get_router())
このコードは、FastAPI を使用して電話会話の要約を取り扱うサーバーを作成しています。EventsManager クラスをカスタマイズして、要約の完了イベントを処理できるようにしています。
EventsManager クラスは、events_manager.EventsManager クラスを継承し、要約の完了イベントを処理します。EventType.TRANSCRIPT_COMPLETE イベントタイプのサブスクリプションを持っています。
handle_event メソッドは、要約の完了イベントを受け取り、add_transcript 関数を使用して要約を保存します。
FastAPI インスタンスが作成され、ドキュメントURLが無効化されています。
ロガーが設定され、デバッグレベルに設定されています。
環境変数から TELEPHONY_SERVER_BASE_URL を取得しています。
TelephonyServer インスタンスが作成され、カスタムの EventsManager インスタンスが含まれています。これにより、要約の完了イベントが処理され、要約が保存されます。
FastAPI インスタンスにルーターが含まれており、TelephonyServer から取得されます。
このサーバーは、電話会話の要約を処理し、保存する機能を提供します。また、FastAPI フレームワークを使用しているため、他のエンドポイントを追加して機能を拡張することが容易です。要約の完了イベントが発生すると、カスタムの EventsManager がイベントを処理し、要約を保存します。
Discussion