How to Build an Advanced End-to-End Voice AI Agent Using Hugging Face Pipelines?

!pip -q set up "transformers>=4.42.0" speed up torchaudio sentencepiece gradio soundfile


import os, torch, tempfile, numpy as np
import gradio as gr
from transformers import pipeline, AutoTokenizer, AutoModelForSeq2SeqLM


DEVICE = 0 if torch.cuda.is_available() else -1


asr = pipeline(
   "automatic-speech-recognition",
   mannequin="openai/whisper-small.en",
   gadget=DEVICE,
   chunk_length_s=30,
   return_timestamps=False
)


LLM_MODEL = "google/flan-t5-base"
tok = AutoTokenizer.from_pretrained(LLM_MODEL)
llm = AutoModelForSeq2SeqLM.from_pretrained(LLM_MODEL, device_map="auto")


tts = pipeline("text-to-speech", mannequin="suno/bark-small")

SYSTEM_PROMPT = (
   "You are a useful, concise voice assistant. "
   "Prefer direct, structured solutions. "
   "If the consumer asks for steps or code, use quick bullet factors."
)


def format_dialog(historical past, user_text):
   turns = []
   for u, a in historical past:
       if u: turns.append(f"User: {u}")
       if a: turns.append(f"Assistant: {a}")
   turns.append(f"User: {user_text}")
   immediate = (
       "Instruction:n"
       f"{SYSTEM_PROMPT}nn"
       "Dialog thus far:n" + "n".be a part of(turns) + "nn"
       "Assistant:"
   )
   return immediate

def transcribe(filepath):
   out = asr(filepath)
   textual content = out["text"].strip()
   return textual content


def generate_reply(historical past, user_text, max_new_tokens=256):
   immediate = format_dialog(historical past, user_text)
   inputs = tok(immediate, return_tensors="pt", truncation=True).to(llm.gadget)
   with torch.no_grad():
       ids = llm.generate(
           **inputs,
           max_new_tokens=max_new_tokens,
           temperature=0.7,
           do_sample=True,
           top_p=0.9,
           repetition_penalty=1.05,
       )
   reply = tok.decode(ids[0], skip_special_tokens=True).strip()
   return reply


def synthesize_speech(textual content):
   out = tts(textual content)
   audio = out["audio"]
   sr = out["sampling_rate"]
   audio = np.asarray(audio, dtype=np.float32)
   return (sr, audio)

def clear_history():
   return [], []


def voice_to_voice(mic_file, historical past):
   historical past = historical past or []
   if not mic_file:
       return historical past, None, "Please file one thing!"
   attempt:
       user_text = transcribe(mic_file)
   besides Exception as e:
       return historical past, None, f"ASR error: {e}"


   if not user_text:
       return historical past, None, "Didn't catch that. Try once more?"


   attempt:
       reply = generate_reply(historical past, user_text)
   besides Exception as e:
       return historical past, None, f"LLM error: {e}"


   attempt:
       sr, wav = synthesize_speech(reply)
   besides Exception as e:
       return historical past + [(user_text, reply)], None, f"TTS error: {e}"


   return historical past + [(user_text, reply)], (sr, wav), f"User: {user_text}nAssistant: {reply}"


def text_to_voice(user_text, historical past):
   historical past = historical past or []
   user_text = (user_text or "").strip()
   if not user_text:
       return historical past, None, "Type a message first."
   attempt:
       reply = generate_reply(historical past, user_text)
       sr, wav = synthesize_speech(reply)
   besides Exception as e:
       return historical past, None, f"Error: {e}"
   return historical past + [(user_text, reply)], (sr, wav), f"User: {user_text}nAssistant: {reply}"


def export_chat(historical past):
   strains = []
   for u, a in historical past or []:
       strains += [f"User: {u}", f"Assistant: {a}", ""]
   textual content = "n".be a part of(strains).strip() or "No dialog but."
   with tempfile.NamedTemporaryFile(delete=False, suffix=".txt", mode="w") as f:
       f.write(textual content)
       path = f.title
   return path

with gr.Blocks(title="Advanced Voice AI Agent (HF Pipelines)") as demo:
   gr.Markdown(
       "##  Advanced Voice AI Agent (Hugging Face Pipelines Only)n"
       "- **ASR**: openai/whisper-small.enn"
       "- **LLM**: google/flan-t5-basen"
       "- **TTS**: suno/bark-smalln"
       "Speak or sort; the agent replies with voice + textual content."
   )


   with gr.Row():
       with gr.Column(scale=1):
           mic = gr.Audio(sources=["microphone"], sort="filepath", label="Record")
           say_btn = gr.Button(" Speak")
           text_in = gr.Textbox(label="Or sort as an alternative", placeholder="Ask me something…")
           text_btn = gr.Button(" Send")
           export_btn = gr.Button(" Export Chat (.txt)")
           reset_btn = gr.Button(" Reset")
       with gr.Column(scale=1):
           audio_out = gr.Audio(label="Assistant Voice", autoplay=True)
           transcript = gr.Textbox(label="Transcript", strains=6)
           chat = gr.Chatbot(peak=360)
   state = gr.State([])


   def update_chat(historical past):
       return [(u, a) for u, a in (history or [])]


   say_btn.click on(voice_to_voice, [mic, state], [state, audio_out, transcript]).then(
       update_chat, inputs=state, outputs=chat
   )
   text_btn.click on(text_to_voice, [text_in, state], [state, audio_out, transcript]).then(
       update_chat, inputs=state, outputs=chat
   )
   reset_btn.click on(clear_history, None, [chat, state])
   export_btn.click on(export_chat, state, gr.File(label="Download chat.txt"))


demo.launch(debug=False)