Why?
, however I don’t prefer it. Why? It fails to do something extra difficult than fundamental voice instructions.
I find yourself utilizing it for 3 issues:
- Get the present date or time
- Get climate info for right now
- Activate or off related units (e.g. TV, lights, robotic vacuum)
that are the one issues that I can use it for reliably. Anything, I get a well mannered and unhelpful “I can’t assist with that”.
Given the rise of LLM Brokers and MCP servers, it’s grow to be simpler than ever to create private assistants and chatbots. And I ask myself,
“Why cease at a chatbot? Why not take this one step additional and create my very own voice assistant?”
That is my try and just do that.
Objectives
So I believe, what precisely do I would like my voice assistant to have the ability to do?
That is my checklist of preliminary objectives:
1. Run on my native laptop
I don’t need to pay for a subscription to make use of an LLM, and in reality, I don’t need to pay for something.
Every part I construct ought to simply run on my native laptop with out having to fret about prices or how a lot free credit score I’ve left on the finish of every month.
2. Replicate Alexa performance
Let’s take child steps — first I merely need to replicate the performance I have already got with Alexa. This shall be a superb milestone to work in the direction of, earlier than I add extra advanced, extravagant options.
It ought to be capable of:
- Get the present date or time
- Get climate info for right now
- Activate or off related units (e.g. TV, lights, robotic vacuum)
earlier than we begin constructing this out right into a fully-fledged Tony Stark’s Jarvis-esque voice assistant that may compute the right way to journey again in time.
3. Be fast
If the responses aren’t quick sufficient, the voice assistant is pretty much as good as being silent.
Asking a query and ready over a minute for a response is unacceptable. I would like to have the ability to ask a query and get a response in an affordable period of time.
Nonetheless, I do know that operating something domestically on my cute little Macbook Air goes to be gradual, no matter what number of tweaks and refactorings I do.
So for now, I’m not going to anticipate millisecond-level response instances. As a substitute the response instances must be faster than the time it takes me to execute the duty/question myself. Not less than on this approach I do know that I’m saving time.
In future articles, we’ll delve deeper into the optimisations I do to get this right down to millisecond response instances with out paying for subscriptions.
My Gadget Specs
- Gadget: Macbook Air
- Chip: Apple M3
- Reminiscence: 16GB
1. Total Construction
I’ve structured the challenge as follows:
Voice Assistant
1. Speech-to-Textual content & Textual content-to-speech
We make use of RealtimeSTT for wakeword detection (e.g. “Alexa”, “Hey Jarvis”, “Hey Siri”), speech detection and real-time speech-to-text transcription.
The transcribed textual content is then despatched to the Agent for processing, after which its response is then streamed to a Kokoro text-to-speech mannequin. The output is then despatched to the speaker.
2. Agent
We use Ollama to run LLMs domestically. The agent and the workflow that it takes is applied in LangGraph.
The agent is chargeable for taking a person question, perceive it, and name on the instruments it thinks are required to supply an applicable response.
Our voice assistant would require the next instruments to satisfy our objectives:
- A operate to get the present date.
- A operate to get the present time.
It additionally wants instruments to work together with smart-home units, however the implementation for this will get fairly concerned so we implement this in a separate MCP server.
3. MCP Server for smart-home Connection
The MCP server is the place we encapsulate the complexity of discovering, connecting to, and managing the units.
A SQL database retains observe of units, their connection info and their names.
In the meantime, instruments are the interface by means of which an agent finds the connection info for a given gadget, after which makes use of it to show the gadget on or off.
Let’s now dive deeper into the implementation particulars of every part.
Need entry to the code repository?
For these of you who want to get entry to the voice-assistant code that accompanies this text, try my Patreon web page right here to get entry PLUS unique entry to neighborhood chats the place you possibly can discuss straight with me about this challenge.
2. Implementation Particulars
Textual content-to-speech (TTS) Implementation
The text-to-speech layer was maybe the simplest to implement.
Given some string we assume comes from the agent, move it by means of a pre-trained text-to-speech mannequin and stream it to the gadget speaker.
Firstly, let’s outline a category referred to as Voice that shall be chargeable for this.
We all know upfront that other than the mannequin that we use for speech synthesis, receiving textual content and streaming it to the speaker would be the identical and might stay decoupled from something mannequin associated.
class Voice():
def __init__(
self,
sample_rate: int = 24000,
chunk_size: int = 2048
):
self.sample_rate = sample_rate
self.chunk_size = chunk_size
self.initialise_model()
def initialise_model(self):
"""Initialise the mannequin to make use of for TTS."""
move
def convert_text_to_speech(self, textual content:str) -> checklist[np.ndarray]:
"""Convert textual content to sepeech and return the waveform as frames."""
move
def communicate(self, textual content:str):
"""Communicate the supplied textual content by means of gadget output."""
frames = self.convert_text_to_speech(self, textual content)
for body in frames:
self.output_stream.write(body.tobytes())so we are able to implement the communicate operate to stream the textual content to the speaker upfront.
Now, we are able to determine which mannequin is on the market, which one to make use of, and the right way to use it, after which wire it up into our Voice class.
TTS Fashions Testing
Under, I checklist the assorted completely different TTS fashions that I experimented with, and the code you need to use to duplicate the outcomes.
1. BarkModel (Hyperlink)
Quickstart code to run the mannequin your self:
from IPython.show import Audio
from transformers import BarkModel, BarkProcessor
mannequin = BarkModel.from_pretrained("suno/bark-small")
processor = BarkProcessor.from_pretrained("suno/bark-small")
sampling_rate = mannequin.generation_config.sample_rate
input_msg = "The time is 3:10 PM."
inputs = processor(input_msg, voice_preset="v2/en_speaker_2")
speech_output = mannequin.generate(**inputs).cpu().numpy()
Audio(speech_output[0], price=sampling_rate)Abstract
- Good: Very reasonable voice synthesis with pure sounding ‘umm’, ‘ahh’ filler phrases.
- Dangerous: High quality is worse with shorter sentences. The tip of the sentence is spoken as if a comply with up sentence will shortly comply with.
- Dangerous: Very gradual. Takes 13 seconds to generate the speech for “The time is 3:10 PM.”
2. Coqui TTS (Hyperlink)
Set up utilizing:
pip set up coqui-ttsTake a look at code
from IPython.show import Audio
from TTS.api import TTS
tts = TTS(model_name="tts_models/en/ljspeech/tacotron2-DDC", progress_bar=False)
output_path = "output.wav"
input_msg = "The time is 3:10 PM."
tts.tts_to_file(textual content=input_msg, file_path=output_path)
Audio(output_path)Abstract
- Good: Quick. Takes 0.3 seconds to generate the speech for “The time is 3:10 PM.”
- Dangerous: Textual content normalisation is less than scratch. In the case of time associated queries, the pronunciation of “PM” is off. When the time is ready to “13:10 PM”, the pronunciation of “13” is unrecognisable.
3. Elevenlabs (Hyperlink)
Set up utilizing:
pip set up elevenlabsand run utilizing:
import dotenv
from elevenlabs.shopper import ElevenLabs
from elevenlabs import stream
dotenv.load_dotenv()
api_key = os.getenv('elevenlabs_apikey')
elevenlabs = ElevenLabs(
api_key=api_key,
)
audio_stream = elevenlabs.text_to_speech.stream(
textual content="The time is 03:47AM",
voice_id="JBFqnCBsd6RMkjVDRZzb",
model_id="eleven_flash_v2_5"
)
stream(audio_stream)Abstract
By far the most effective by way of high quality and response instances, which clearly it must be given it’s a paid service.
Additionally they present some free credit with no subscription, however I’d reasonably not grow to be depending on it in any respect when growing my voice assistant so we skip it for now.
4. Kokoro (Hyperlink)
We go away the most effective til final.
Set up utilizing:
pip set up kokoro pyaudioTake a look at code:
RATE = 24000
CHUNK_SIZE = 1024
p = pyaudio.PyAudio()
print(f"Enter gadget: {p.get_default_input_device_info()}")
print(f"Output gadget: {p.get_default_output_device_info()}")
output_stream = p.open(
format=pyaudio.paFloat32,
channels=1,
price=RATE,
output=True,
)
input_msg = "The time is 03:47AM"
generator = pipeline(input_msg, voice='af_heart')
for i, (gs, ps, audio) in enumerate(generator):
print(i, gs, ps)
for begin in vary(0, len(audio), CHUNK_SIZE):
chunk = audio[start:start + CHUNK_SIZE]
output_stream.write(chunk.numpy().astype(np.float32).tobytes())Abstract
Firstly, it’s fast — it’s on par with Elevenlabs, solely marginally slower, probably not noticeable given the instance textual content.
Secondly, the standard of the speech can be good. Positive, it might be higher, there are events the place it sounds barely clunky.
However on common the standard of the outputs are spot on.
Defining the Voice Class
So, we resolve to make use of Kokoro for our text-to-speech implementation. Let’s now fill within the blanks for our Voice class. Additionally take into account that this can be a first implementation, and I do know sooner or later I’ll need to attempt different fashions.
So as a substitute of implementing the mannequin particular code straight into the Voice class, I’ll create a baby class that inherits from Voice.
This fashion, I can do a fast switcharoo between completely different fashions with out having to alter the Voice class or unravel code that has grow to be coupled.
from kokoro import KPipeline
class KokoroVoice(Voice):
def __init__(self, voice:str, sample_rate: int = 24000, chunk_size: int = 2048):
"""Initialise the mannequin to make use of for TTS.
Args:
voice (str):
The voice to make use of.
See https://github.com/hexgrad/kokoro/blob/predominant/kokoro.js/voices/
for all voices.
sample_rate (int, optionally available):
The pattern price to make use of. Defaults to 24000.
chunk_size (int, optionally available):
The chunk measurement to make use of. Defaults to 2048.
"""
self.voice = voice
tremendous().__init__(sample_rate, chunk_size)
def initialise_model(self):
"""Load the mannequin to make use of for TTS."""
self.pipeline = KPipeline(lang_code="b")
def convert_text_to_speech(self, textual content:str) -> checklist[np.ndarray]:
"""Convert textual content to speech and return the waveform as frames."""
generator = self.pipeline(textual content, voice=self.voice)
frames = []
for i, (_, _, audio) in enumerate(generator):
for begin in vary(0, len(audio), self.chunk_size):
chunk = audio[start : start + self.chunk_size]
frames.append(chunk.numpy().astype(np.float32))
return framesNow, this implementation permits us to easily import and instantiate this class on the level the place we obtain textual content from the agent, and stream it to the gadget speaker utilizing:
textual content = "Hiya world"
voice = KokoroVoice(**kwargs)
voice.communicate(textual content)SmartHome MCP Server Implementation
This MCP server is devoted to discovering, connecting and managing smarthome units. It lives in a separate repository, properly separated from the voice assistant.
On the time of writing, the one smarthome gadget I’ve is a Tapo Sensible Plug. You possibly can work together with Tapo units through the use of the python-kasa library.
Our server must do the next:
- Given a tool title, flip it on or off.
- Uncover new units and add them to the database.
- Replace the gadget database with the newest gadget info — this consists of the title of the gadget, the IP tackle and the MAC tackle.
1. Database
Firstly, let’s take a look at how we are going to retailer the gadget info in a SQL database. For simplicity I’ll select duckdb because the database backend.
Units Desk
We firstly outline the schema for our first (and solely) desk referred to as gadget.
# src/smarthome_mcp_server/database.py
import os
import duckdb
from dataclasses import dataclass
@dataclass
class TableSchema:
title:str
columns:dict[str, str]
primary_key:checklist[str]
def get_device_table_schema():
return TableSchema(
title="gadget",
columns={
"device_id" : "VARCHAR",
"title": "VARCHAR",
"ip_address": "VARCHAR",
},
primary_key=["device_id"],
)The device_id is the first key, and by definition must uniquely establish all units in our dwelling. Fortunately, every Tapo gadget has a novel device-id that we are able to use.
The title is what the person can be referencing because the gadget title. For instance, in our case, the Tapo Sensible Plug is related to our front room mild, and is called lights. This title is assigned by way of thee Tapo App.
Lastly, the ip_address column would be the IP Tackle that’s used to hook up with the gadget to be able to management it.
DB Initialisation
We create some helper capabilities like get_create_table_if_not_exists_query and initialise_database capabilities that we are able to name to invoke to create the DB on first startup.
For brevity, I present simply the initialise_database operate because the former is self-explanatory:
def initialise_database(db_path:os.PathLike) -> duckdb.DuckDBPyConnection:
"""Get the database connection and create the tables if they do not exist."""
conn = duckdb.join(db_path)
# initialise if not exists tables
conn.execute(
get_create_table_if_not_exists_query(get_device_table_schema())
)
return connGadget administration
Lastly, we implement the code that shall be interacting with the units and updating the database.
import duckdb
from dotenv import
class DeviceManager:
def __init__(self, conn:duckdb.DuckDBPyConnection) -> None:
self._conn = conn
...
async def turn_on_device(self, device_name: str) -> str:
"""Activate a tool.
Args:
device_name (str):
The title of the gadget to activate.
"""
attempt:
gadget = await self._get_device(device_name)
besides DeviceNotFoundError as e:
logger.exception(e)
return f"Gadget {device_name} not discovered."
await gadget.turn_on()
return f"Gadget {device_name} turned on."
async def turn_off_device(self, device_name: str) -> str:
"""Flip off a tool.
Args:
device_name (str):
The title of the gadget to show off.
"""
attempt:
gadget = await self._get_device(device_name)
besides DeviceNotFoundError as e:
logger.exception(e)
return f"Gadget {device_name} not discovered."
await gadget.turn_off()
return f"Gadget {device_name} turned off."
async def list_devices(self) -> checklist[str]:
"""Record the out there gadget names.
Returns:
checklist[str]:
An inventory of gadget names.
"""
outcomes = self._conn.question("SELECT title FROM gadget").fetchall()
return [result[0] for end in outcomes]The three strategies above would be the public strategies that we register as instruments for our Voice Assistant.
We’ve omitted the non-public strategies for brevity.
One factor that I’ve realised since scripting this code is that DeviceManager could be very Tapo particular. After having checked out integrating non-Tapo units, I realised I’ve been naive to suppose that different sensible gadget APIs would comply with the identical, standardised sample.
So sooner or later, this class will must be modified to TapoDeviceManager, and additional abstractions will must be made to accommodate this variability.
For instance, just lately I’ve bought some Wiz lightbulbs for my bed room. Seems, the API doesn’t fetch the names assigned to every gadget by way of the app, which was out there in Tapo by default.
Due to this fact, I might want to consider some option to fetch this within the backend, or use the voice-assistant to populate it when it doesn’t exist.
3. Expose the instruments to Voice-Assistant utilizing FastMCP
Lastly, we have to expose the strategies we’ve written as instruments for our voice assistant to make use of.
from fastmcp import FastMCP
def register_device_manager_tools(mcp_instance: FastMCP, device_manager: DeviceManager) -> FastMCP:
"""Register the strategies outlined in DeviceManager as instruments for MCP server."""
mcp_instance.device(name_or_fn=device_manager.list_devices)
mcp_instance.device(name_or_fn=device_manager.turn_off_device)
mcp_instance.device(name_or_fn=device_manager.turn_on_device)
return mcp_instance
async def populate_database(device_manager: DeviceManager):
"""Discover all units which can be out there and replace the database.
Uncover all out there units and get their newest states.
Be aware:
Gadget names might have modified by way of the cellular app, thus this
step is critical when beginning the server.
"""
all_devices = await device_manager.discover_new_devices()
upsert_coroutines = [device_manager._upsert_device(device) for device in all_devices.values()]
await asyncio.collect(*upsert_coroutines)
def initialise_server(db_path: os.PathLike) -> FastMCP:
"""Initialise the server.
Args:
db_path (os.PathLike):
The trail to the duckdb database which
shops the server info.
Returns:
FastMCP: The FastMCP server.
"""
conn = initialise_database(db_path)
device_manager = DeviceManager(conn)
# discover all units which can be out there and replace the database
asyncio.run(populate_database(device_manager))
mcp = FastMCP(
title="smarthome-mcp-server",
directions="This server is for locating and controlling smarthome units.",
)
register_device_manager_tools(mcp, device_manager)
return mcpinitialise_server is the place we initialise and pre-populate the database, and run the server.
Discover we populate the database on startup every time. That is mandatory since gadget names may have been up to date by way of the Tapo app between runs, so that is an try and fetch essentially the most up-to-date info for all units.
Now, I do know there are some holes within the implementation — it’s a primary try and an ongoing challenge, so should you see any points or potential enhancements please let me know by way of my Patreon account (see finish of article).
Server Entry Level
We use typer to make our server right into a CLI app.
# __main__.py
load_dotenv()
app = typer.Typer()
console = Console()
@app.command()
def predominant():
config = load_config()
# arrange server information listing
root_dir = platformdirs.user_data_path(
appname="smarthome-mcp-server",
ensure_exists=True
)
db_path = Path(root_dir) / config.database.path
db_path.dad or mum.mkdir(mother and father=True, exist_ok=True)
logger.information("Server information listing: %s", db_path)
# init and run
mcp_instance = initialise_server(db_path)
asyncio.run(mcp_instance.run_stdio_async())
if __name__ == "__main__":
app()We then run the server python3 -m smarthome_mcp_server:
╭─ FastMCP 2.0 ────────────────────────────────────────────────────────────╮
│ │
│ _ __ ___ ______ __ __ _____________ ____ │
│ ____ │
│ _ __ ___ / ____/___ ______/ /_/ |/ / ____/ __ |___ / __ │
│ │
│ _ __ ___ / /_ / __ `/ ___/ __/ /|_/ / / / /_/ / ___/ / / / / │
│ / │
│ _ __ ___ / __/ / /_/ (__ ) /_/ / / / /___/ ____/ / __/_/ /_/ │
│ / │
│ _ __ ___ /_/ __,_/____/__/_/ /_/____/_/ /_____(_)____/ │
│ │
│ │
│ │
│ 🖥️ Server title: smarthome-mcp-server │
│ 📦 Transport: STDIO │
│ │
│ 📚 Docs: https://gofastmcp.com │
│ 🚀 Deploy: https://fastmcp.cloud │
│ │
│ 🏎️ FastMCP model: 2.11.2 │
│ 🤝 MCP model: 1.12.4 │
│ │
╰──────────────────────────────────────────────────────────────────────────╯
[08/19/25 05:02:55] INFO Beginning MCP server server.py:1445
'smarthome-mcp-server' with
transport 'stdio' 4. Utilizing the SmartHome Instruments
Now that the server has been applied, we are able to now outline some strategies that may work together with the server by way of a shopper. This shopper shall be used to register the instruments for the Voice Assistant to make use of.
Coming again to the voice-assistant repo:
from langchain_mcp_adapters.shopper import MultiServerMCPClient
def get_new_mcp_client() -> MultiServerMCPClient
return MultiServerMCPClient(
{
"smarthome-mcp-server": {
"command": "smarthome_mcp_server",
"args": [],
"transport": "stdio",
}
}
)This methodology makes use of the handy MultiServerMCPClient class to register our smarthome MCP server for device utilization.
The returned shopper object then exposes a get_tools methodology which returns all of the instruments that the registered servers expose.
mcp_client = get_new_mcp_client()
instruments = await mcp_client.get_tools()Be aware how we use await right here given the get_tools methodology is asynchronous.
By defining a operate referred to as get_mcp_server_tools:
def get_mcp_server_tools():
mcp_client = get_new_mcp_client()
instruments = await mcp_client.get_tools()
return instrumentsthis single operate will be imported into wherever we outline our agent and register the instruments to be used.
Speech-to-text Implementation
Speech-to-text (STT) is the place numerous complexity is available in because it requires realtime IO processing.
STT itself is easy sufficient to attain — there are many fashions on the market that we are able to use. However what makes it advanced is the necessity to have the ability to always pay attention for a person’s voice enter, which consists of a wakeword and a question.
A wakeword is what you usually use to set off a voice assistant to start out listening to you. For instance, “Hey Google” or “Hey Siri”, or “Alexa”.
I may write this code fully myself, however to make issues easier, I had a fast dig round simply in case there was one thing pre-built that I may use.
And to my shock, I discovered the package deal RealtimeSTT (hyperlink right here) and it really works completely.
The way it works in a nutshell
- Create a thread for listening to the person’s voice enter. One other for transcribing, which runs the STT mannequin.
- If a wakeword is detected, begin recording the person’s voice enter.
- The recorded audio is then despatched to the STT mannequin for transcribing, and returns the transcribed textual content as a string.
To make use of this package deal, all we have to do is use the AudioToTextRecorder class as a context supervisor like under:
from RealtimeSTT import AudioToTextRecorder
with AudioToTextRecorder(
mannequin='tiny',
wakeword_backend='oww',
wake_words='hey jarvis',
gadget='cpu',
wake_word_activation_delay=3.0,
wake_word_buffer_duration=0.15,
post_speech_silence_duration=1.0
) as recorder:
whereas True:
# get the transcribed textual content from recorder
question = recorder.textual content()
if (question will not be None) and (question != ""):
# get response from our langgraph agent
response_stream = await get_response_stream(
question, agent_executor, thread_config
)
# output the response to gadget audio
await stream_voice(response_stream, output_chunk_builder, voice)We’ll come again to get_response_stream and stream_voice strategies within the subsequent part, since this additionally includes how we outline our agent.
However merely placing collectively the AudioToTextRecorder context supervisor in the best way now we have, we’ve bought a working speech -> textual content -> response mechanism applied.
For those who had been to easily substitute the get_response_stream with any LLM agent, and substitute the stream_voice with any text-to-speech agent, you’ll have a working voice assistant.
You would additionally use a easy print assertion and you’ll have a rudimentary chat bot with voice enter.
Agent Implementation
Lastly, the good things — the agent implementation.
I’ve left this as final because it’s a bit extra concerned. Let’s get caught in.
LangGraph — What’s it?
LangGraph is a framework for constructing stateful, graph-based workflows with language mannequin brokers.
Nodes encapsulate any logic associated to an motion an LLM agent can take.
Edges encapsulate the logic which determines the right way to transition from one node to a different.
LangGraph implements a prebuilt graph that we are able to get by way of the create_react_agent methodology. The graph appears like this:
create_react_agent methodologyLet’s use this for instance to elucidate higher how nodes and edges work.
As you possibly can see, the graph could be very easy:
- Given a question (the
__start__node) - The
agentnode will obtain the question and decide whether or not it must name a device to have the ability to reply appropriately.- If it does, we transition to the
devicenode. As soon as the device response is obtained, we return to theagentnode. - The agent will repeatedly name the suitable instruments till it determines it has every thing it wants.
- If it does, we transition to the
- Then, it would return its response (the
__end__node)
The conditional transition between the agent, instruments and __end__ node is represented as dashed strains. Then, the query is:
How will we decide which node to go to subsequent?
Properly, Langgraph maintains a log of the messages which have been despatched, and this represents the state of the graph.
The messages can come from the person, the agent, or a device. On this instance, the agent node will create a message that explicitly states that it’s going to name a device (precisely how shall be revealed within the subsequent part).
The presence of this device name is what triggers the transition from the agent node to the instruments node.
If no instruments are referred to as, then the transition from the agent node to the __end__ node is triggered.
It’s this test for the presence of device calls that’s applied within the conditional edge between the agent, instruments and __end__ nodes.
In a future article, I’ll go into an instance of how I created a customized agent graph to optimise for latency, and reveal how precisely these conditional edges and nodes are applied.
For now, we don’t want to enter an excessive amount of element about this because the prebuilt graph is sweet sufficient for the scope of this text.
Our Agent Implementation
So, we outline a operate referred to as get_new_agent like under:
from langgraph.prebuilt import create_react_agent
from langgraph.graph.state import CompiledStateGraph
from voice_assistant.instruments.datetime import get_tools as get_datetime_tools
def get_new_agent(
config, short_term_memory, long_term_memory
) -> CompiledStateGraph:
"""Construct and return a brand new graph that defines the agent workflow."""
# initialise the LLM
mannequin = init_chat_model(
mannequin=config.Agent.mannequin,
model_provider=config.Agent.model_provider,
temperature=0,
reasoning=config.Agent.reasoning
)
# initialise the instruments that the agent will use
server_tools = await get_mcp_server_tools()
instruments = (
get_datetime_tools()
+ server_tools
)
# construct the agent workflow given the LLM, its instruments and reminiscence.
agent_executor = create_react_agent(
mannequin,
instruments,
checkpointer=short_term_memory,
retailer=long_term_memory
)
return agent_executorwhich is chargeable for:
- Initialising the LLM
init_chat_modelreturns the LLM from the required supplier. In our case, we use Ollama as our supplier, andllama3.2:newestas our mannequin kind.
- Defining the total set of instruments that the agent will use.
- We’ve a operate referred to as
get_datetime_tools()which returns achecklistofStructuredToolobjects. - We even have
server_tools, that are the checklist of instruments that our beforehand talked about MCP server gives for dwelling automation. - Moreover, If we want to prolong the set of instruments the agent can use, that is the place so as to add them.
- We’ve a operate referred to as
- Assemble the agent workflow given the LLM and its instruments.
- Right here we name the
create_react_agentoperate from LangGraph. - The operate may absorb
checkpointerandretailerobjects that are used to persist the state of the agent, appearing as a brief time period and long run reminiscence. - Sooner or later, if we need to use a customized graph, we are able to substitute the
create_react_agentoperate name with our personal implementation.
- Right here we name the
Dealing with the Agent Response
Now, we’ve thus far applied all of the parts that we have to
- Get the person question
- Get the instruments
- Create the agent
The following step is to run the agent to get a response for the question, and output it by way of the Voice methodology we outlined earlier.
Given the person question textual content that we’ve obtained from our STT implementation, we format it right into a dictionary:
user_query = "Hiya world!"
user_query_formatted = {
"position": "person",
"content material": user_query
}This dictionary tells the agent that the message is from the person.
We additionally add a system immediate to set the context and provides directions to the agent:
system_prompt_formatted = {
"position": "system",
"content material": (
"You're a voice assistant referred to as Jarvis."
+ " Preserve your responses as brief as doable."
+ "Don't format your responses utilizing markdown, resembling **daring** or _italics. ",
)
}These two messages are then handed into the agent to get a response:
response = agent_executor.invoke(
{"messages" : [system_prompt_formatted, user_query_formatted]},
)The response is a dictionary of messages (for brevity we omit any superfluous content material):
output
> {
"messages": [
SystemMessage(
content="You are a voice assistant called Jarvis.Keep your responses as short as possible.Do not format your responses using markdown, such as **bold** or _italics. ",
additional_kwargs={},
...
),
HumanMessage(
content="What time is it?",
additional_kwargs={},
...
),
AIMessage(
content="",
additional_kwargs={},
tool_calls=[
{
"name": "get_current_time",
"args": {},
"id": "b39f7b12-4fba-494a-914a-9d4eaf3dc7d1",
"type": "tool_call",
}
],
...
),
ToolMessage(
content material="11:32PM",
title="get_current_time",
...
),
AIMessage(
content material="It is presently 11:32 PM.",
additional_kwargs={},
...
),
]
}As you possibly can see, the output is an inventory of all of the messages which have been created all through the graph execution.
The primary message will all the time be a HumanMessage or a SystemMessage since that is what we supplied to the agent as enter (i.e. the __start__ node).
The remaining are the messages that the agent or instruments returned, within the order they had been referred to as.
For instance, you possibly can see the primary AIMessage, the message kind generated by the LLM, has a device name inside it which makes use of a get_current_time device.
The presence of a tool_calls property within the AIMessage is what triggers the conditional transition from the agent node to the instruments node.
agent and instruments highlighted in purple.You then see the ToolMessage which is the response that was returned by the get_current_time device.
Lastly, the mannequin responds with the precise response to the person question. The shortage of a tool_calls property within the AIMessage implies that the graph ought to transition to the __end__ node and return the response.
Decreasing Latency
Coming again to invoking the agent to get a response, the difficulty with utilizing the invoke methodology is that we anticipate the complete workflow to finish earlier than we get a response.
This could take a very long time, particularly if the agent is addressing a fancy question. In the meantime, the person is ready idly for the agent to reply, which ends up in a poor person expertise.
So to enhance on this, we are able to use the stream mode in LangGraph to stream the response as they’re generated.
This permits us to start out voicing the response as they arrive, reasonably than ready for the complete response to be generated after which voicing it multi function go.
output_stream = agent_executor.stream(
{"messages" : [system_prompt_formatted, user_query_formatted]},
stream_mode="messages"
)Right here, output_stream is a generator that may yield a tuple of messages and message metadata, as they arrive.
Be aware, there’s an asynchronous model of this methodology referred to as astream, which does precisely the identical factor however returns an AsyncIterator as a substitute.
If we take a look at the messages we get after this modification:
print([chunk for chunk, metadata in output])
> AIMessageChunk(
content material="",
tool_calls=[{"name": "get_current_time", ...}],
tool_call_chunks=[{"name": "get_current_time", "args": "{}", ...}],
),
ToolMessage(content material="01:21AM", title="get_current_time", ...),
AIMessageChunk(content material="It", ...),
AIMessageChunk(content material="'s", additional_kwargs={}, ...),
AIMessageChunk(content material=" presently", ...),
AIMessageChunk(content material=" ",),
AIMessageChunk(content material="1", ...),
AIMessageChunk(content material=":", ...),
AIMessageChunk(content material="21", ...),
AIMessageChunk(content material=" AM", ...),
AIMessageChunk(content material=".", ...),
AIMessageChunk(content material="", ...),Now you can see the tokens are being returned as they’re generated.
However this poses a brand new downside!
We will’t simply give the TTS mannequin particular person tokens, since it would simply pronounce every token one after the other, i.e. "It", "'s" shall be pronounced individually, which is certainly not what we would like.
So, there’s a tradeoff that we have to make
Whereas we have to stream the response to minimise person wait time, will nonetheless want to attend to build up sufficient tokens that kind a significant chunk, earlier than sending them to the TTS mannequin.
Constructing Output Chunks
We due to this fact deal with this complexity by defining an OutputChunkBuilder. So what constitutes a significant chunk?
The very first thing that involves thoughts is to attend till a full sentence, i.e. append all of the tokens till it ends with one among ., ?, ;, !.
From trial and error, it has additionally confirmed smart to incorporate n on this checklist as nicely, once we get a very lengthy response from the agent that makes use of bullet factors.
class OutputChunkBuilder:
def __init__(self):
self._msg = ""
self.end_of_sentence = (".", "?", ";", "!", "n")
def add_chunk(self, message_chunk:str):
self._msg += message_chunk
def output_chunk_ready(self) -> bool:
return self._msg.endswith(self.end_of_sentence)We obtain this with the above code, consisting of 1 operate that appends message chunks collectively right into a buffer referred to as _msg, and one to test if the collated messages are prepared (i.e. is it a full sentence or does it finish with a brand new line).
class OutputChunkBuilder:
... # omitted for brevity
def _reset_message(self):
self._msg = ""
def get_output_chunk(self):
msg = self._msg # Get the present message chunk
self._reset_message()
return msgWe additionally implement the get_output_chunk operate which can return the messages collated thus far, and in addition reset the message buffer to an empty string in order that it’s prepared for collating the following set of chunks.
This allows us to make use of logic like under to stream the response, sentence by sentence:
def stream_voice(msg_stream, output_chunk_builder):
for chunk, metadata in msg_stream:
# append the chunk to our buffer
if chunk.content material != "":
output_chunk_builder.add_chunk(chunk.content material)
# communicate the output chunk whether it is prepared
if output_chunk_builder.output_chunk_ready():
voice.communicate(output_chunk_builder.get_output_chunk())Instruments Implementation
Lastly, let’s take a look at how we are able to implement the instruments required to get the present date and time.
That is very simple, by far the simplest implementation. Any operate that you simply create can be utilized as a device so long as the docstrings are well-written and formatted clearly.
There are two predominant methods to mark a operate as a device:
- Utilizing the
@devicedecorator fromlangchain_core.instruments - Utilizing the
StructuredToolclass fromlangchain_core.instruments.structured
For simpler unit testing of our instruments, we go for the second possibility because the first possibility doesn’t permit us to import the device operate into our assessments.
First, write the capabilities to get the time and date as we might do usually:
# instruments/datetime.py
from datetime import datetime
from langchain_core.instruments.structured import StructuredTool
def get_now_datetime() -> datetime:
"""Wrapper for simpler mocking in unit take a look at."""
return datetime.now()
def get_current_time() -> str:
"""Get the present time in format HH:MM AM/PM"""
return get_now_datetime().strftime("%I:%Mpercentp")Moreover, we write a easy wrapper operate referred to as get_now_datetime that returns the present datetime, which additionally makes it simpler to mock in our unit assessments.
Subsequent, a operate for getting the present date.
def _convert_date_to_words(dt: datetime):
"""Change date values represented in YYYY-mm-dd format to phrase values as they'd be pronounced."""
day = dt.day
if day == 1 or day == 21 or day == 31:
day_word = f"{day}st"
elif day == 2 or day == 22:
day_word = f"{day}nd"
elif day == 3 or day == 23:
day_word = f"{day}rd"
else:
day_word = f"{day}th"
date_obj = dt.strftime(f"%B {day_word}, %Y")
return date_obj
def get_current_date() -> str:
"""Get the present date in format YYYY-MM-DD"""
dt = get_now_datetime()
dt_str = _convert_date_to_words(dt)
return dt_strWe’ve to watch out right here — completely different text-to-speech (TTS) fashions have various skills with regards to textual content normalisation.
Instance
If the operate
get_current_datereturns the string01-01-2025, the TTS mannequin might pronounce this as ‘oh one oh one twenty twenty 5‘.
To make our implementation sturdy to such variations, we normalise the date string to be clearer in how the date must be pronounced utilizing the _convert_date_to_words operate.
In doing so, we convert a datetime object like datetime(2025, 1, 1) into January 1st, 2025.
Lastly, we write a get_tools operate which can wrap up the get_current_time and get_current_date strategies right into a StructuredTool, and return them in an inventory:
def get_tools():
"""Get an inventory of instruments for the agent.
Returns:
An inventory of device capabilities out there to the agent.
"""
return [
StructuredTool.from_function(get_current_time),
StructuredTool.from_function(get_current_date),
]thereby permitting us to import this operate and callling it once we create the agent, as we noticed within the agent implementation part.
Placing all of it collectively to construct our Agent
Now, we’ve gone by means of the person parts that make up our voice assistant, time to assemble them collectively.
# predominant.py
from RealtimeSTT import AudioToTextRecorder
from langgraph.checkpoint.sqlite.aio import AsyncSqliteSaver
from langgraph.retailer.sqlite.aio import AsyncSqliteStore
from voice_assistant.agent import get_new_agent, get_response_stream
from voice_assistant.voice import KokoroVoice
from settings import load_config
async def predominant():
conf = load_config()
voice = KokoroVoice(**conf.KokoroVoice)
output_chunk_builder = OutputChunkBuilder()
thread_config = {"configurable": {"thread_id": "abc123"}}
# brief time period reminiscence
async with AsyncSqliteSaver.from_conn_string(conf.Agent.reminiscence.checkpointer) as saver:
# long run reminiscence
async with AsyncSqliteStore.from_conn_string(conf.Agent.reminiscence.retailer) as retailer:
agent_executor = await get_new_agent(conf, saver, retailer)
with AudioToTextRecorder(**conf.AudioToTextRecorder) as recorder:
whereas True:
question = recorder.textual content()
if (question will not be None) and (question != ""):
response_stream = await get_response_stream(
question, agent_executor, thread_config
)
await stream_voice(response_stream, output_chunk_builder, voice)
if __name__ == "__main__":
asyncio.run(predominant())Firstly, we load in our Yaml config file utilizing OmegaConf (hyperlink right here). The settings module and the load_config implementation is like under:
# settings.py
import logging
from pathlib import Path
from omegaconf import OmegaConf
logger = logging.getLogger(__name__)
CONFIG_PATH = Path(__file__).mother and father[1] / "conf" / "config.yaml"
def load_config():
logger.debug(f"Loading config from: {CONFIG_PATH}")
return OmegaConf.load(CONFIG_PATH)Secondly, we use SQL databases to retailer our brief and long run reminiscence — that is achieved utilizing the AsyncSqliteSaver and AsyncSqliteStore lessons from the checkpoint and retailer modules in langgraph.
from langgraph.checkpoint.sqlite.aio import AsyncSqliteSaver
from langgraph.retailer.sqlite.aio import AsyncSqliteStore
... # omitted for brevity
# brief time period reminiscence
async with AsyncSqliteSaver.from_conn_string(conf.Agent.reminiscence.checkpointer) as saver:
# long run reminiscence
async with AsyncSqliteStore.from_conn_string(conf.Agent.reminiscence.retailer) as retailer:
agent_executor = await get_new_agent(conf, saver, retailer)
... # omitted for brevity
Then, shortly loop, the STT thread information the person’s voice enter after a wakeword is detected, which is then handed to the agent for processing.
The agent response is returned as an AsyncIterator, which we then stream to the gadget audio system utilizing the stream_voice operate.
The stream_voice operate appears like this:
async def stream_voice(
msg_stream: AsyncGenerator,
output_chunk_builder: OutputChunkBuilder,
voice: Voice
):
"""Stream messages from the agent to the voice output."""
async for chunk, metadata in msg_stream:
if metadata["langgraph_node"] == "agent":
# construct up message chunks till a full sentence is obtained.
if chunk.content material != "":
output_chunk_builder.add_chunk(chunk.content material)
if output_chunk_builder.output_chunk_ready():
voice.communicate(output_chunk_builder.get_output_chunk())
# if now we have something left within the buffer, communicate it.
if output_chunk_builder.current_message_length() > 0:
voice.communicate(output_chunk_builder.get_output_chunk())Which is similar logic as we already mentioned earlier than within the Constructing Output Chunks part, however with some small tweaks.
It seems, not all responses finish with a punctuation mark.
For instance, when the LLM makes use of bullet factors of their response, I’ve discovered they omit the punctuation for every bullet level.
So, we ensure that to flush our buffer on the finish if it isn’t empty.
We additionally filter out any messages that aren’t from the agent, as we don’t need to stream the person’s enter or the device responses again to the gadget audio system. We do that by checking the langgraph_node metadata key, and solely talking the message if it comes from the agent.
And seems, that’s all you want to construct a completely functioning voice assistant.
Ultimate Remarks
Total, I’ve been pleasantly shocked at how simple it was to construct this out.
Positive, there are definitely extra optimisations that may be made, however given I’ve been capable of get the total performance working inside two weeks (while working a full-time job), I’m pleased with the outcomes.
However we’re not achieved but.
There are an entire load of issues I couldn’t focus on to cease this text turning into a whole e-book, resembling the extra optimisations I’ve needed to make to make the voice assistant faster, so this shall be lined in my subsequent article.
For these of you who loved this text, try my different articles on Medium, at https://medium.com/@bl3e967
