LangChain Agent를 활용한 음성 인증 봇

import os

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.messages import AIMessage, HumanMessage
from langchain.agents import create_tool_calling_agent
from langchain.agents import AgentExecutor, create_tool_calling_agent, tool
from langchain_community.tools.tavily_search import TavilySearchResults
import speech_recognition as sr
from playsound import playsound
import librosa
import torch
import torch.nn.functional as F
from transformers import Wav2Vec2Model, AutoFeatureExtractor
from torch.nn.functional import cosine_similarity

from dotenv import load_dotenv
load_dotenv()

OPENAI_API_KEY = os.getenv("OPENAI_API_KEY_songhun")

device = torch.device("cuda" if torch.cuda.is_available() else "cpu")

def load_and_process_audio(file_path, feature_extractor, max_length=4.0):
    audio, sampling_rate = librosa.load(file_path, sr=16000)
    inputs = feature_extractor(audio, sampling_rate=sampling_rate, return_tensors="pt", padding="max_length", truncation=True, max_length=int(max_length * sampling_rate))
    return inputs.input_values.to(device)

def get_voice_similarity():
    model_name = "Songhun/wav2vec2-base-960h-contrastive"
    voice_model = Wav2Vec2Model.from_pretrained(model_name).to(device)
    feature_extractor = AutoFeatureExtractor.from_pretrained(model_name)
    audio_input1 = load_and_process_audio('./anchor_voice.wav', feature_extractor).to(device)
    audio_input2 = load_and_process_audio('./saved_voice.wav', feature_extractor).to(device)

    embedding1 = voice_model(audio_input1).last_hidden_state.mean(dim=1)
    embedding2 = voice_model(audio_input2).last_hidden_state.mean(dim=1)

    similarity = F.cosine_similarity(embedding1, embedding2).item()

    voice_return = False
    if similarity > 0.3331:
        voice_return = True
    print(f"Similarity: {similarity}")
    return voice_return, similarity

@tool
def get_voice_and_verify():
    """
    This function is executed when voice authentication is required.
    This function records audio and saves it, and then calculates the similarity to verify if it's the same voice.
    this function return voice_return(True, False), similarity(Voice Similarity Score)
    """
    r = sr.Recognizer()
    mic = sr.Microphone()
    with mic as source:
        print("녹음 시작...")
        audio_data = r.listen(source)
        print("녹음 완료!")

    with open('./saved_voice.wav', "wb") as f:
        f.write(audio_data.get_wav_data())

    return get_voice_similarity()

def main():
    model = ChatOpenAI(model="gpt-4o", openai_api_key=OPENAI_API_KEY)
    prompt = ChatPromptTemplate.from_messages([
        ('system', '너는 유용한 인공지능 조수야'),
        ("placeholder", "{chat_history}"),
        ("human", "{input}"),
        ("placeholder", "{agent_scratchpad}"),
    ])

    tools = [get_voice_and_verify]
    agent = create_tool_calling_agent(model, tools, prompt)
    agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True)
    
    chat_history = []

    while True:
        input_text = input("궁금한걸 물어보세요:")
        if input_text=='exit':break

        chat_history.append(HumanMessage(content=input_text))

        ans = agent_executor.invoke(
            {
                "input": input_text,
                "chat_history": chat_history,
            }
        )

        chat_history.append(AIMessage(content=ans['output']))
        print(f"에이전트: {ans}")
    print('종료합니다.')