FastMCP Voice Stack, parte 2: Orquestación de LangGraph para agentes vocales

De MCP y entrada/salida por voz a un bucle de agente

La primera parte trataba sobre MCP, los servidores de herramientas FastMCP, la conversión de voz a texto y de texto a voz, además de una demostración de voz sin uso de modelos de lenguaje. En esta parte se añaden LangChain y LangGraph para que el asistente pueda razonar, llamar a esas herramientas y mantener memoria a corto plazo, siempre que sea posible mediante niveles de API gratuitos.

LangChain versus LangGraph en una sola página

LangChain conecta modelos (OpenAI, Anthropic, Groq, Ollama y otros) con datos, herramientas y flujos de trabajo. LangGraph es la capa de orquestación para agentes multi-paso, de larga duración y con estado, expresados como grafos.

Un agente realista debe poder pensar, buscar, resumir, escribir archivos y enviar correos. Cada uno de estos pasos corresponde a un nodo, que es una función en Python. Los nodos son los puntos; las aristas son las líneas que los conectan.

a. Search the web for the latest AI news
b. Summarize it
c. Save it to a file
d. Email file

Instale el conjunto de herramientas en el entorno existente de la Parte Uno:

cmd>> uv pip install langgraph langchain langchain_groq python-dotenv

# if using OpenAI API use langchain_openai instead of langchain_groq

Los ayudantes simples de LangChain tienen dificultades con la memoria duradera, las ramificaciones complejas, las reintentos y las pausas que permiten la intervención humana. LangGraph aborda estos problemas mediante estados y conexiones explícitas.

Vocabulario

Gráfico: todo el flujo de trabajo. Nodo: una función de tarea. Conexión: lo que se ejecuta a continuación. Estado: datos compartidos que cada nodo puede leer y actualizar. START / END: punto de entrada y salida. compile() crea un gráfico ejecutable; invoke() lo ejecuta con un estado inicial.

Gráfico de saludo

Un pequeño ejemplo adecuado para cuadernos de notas:

#!/usr/bin/python3
from typing import TypedDict  # we will use this to store our state
from langgraph.graph import StateGraph, START, END


# ---Step 1: Define the state Schema
#- State holds shared data passed between nodes
#- Each node updates the field in our case now it is message
class AgentState(TypedDict):
    # holds our message---> read on pydantic to understand the state better
    message: str

# ----Step 2: Create a node
# - Node a python function that performs one task
# - The node receives the current state we defined above, does some work, and returns the updated state
def greet(state: AgentState)-> AgentState:
    """This simple node adds a greeting message"""
    state['message']='Hey ' + state['message'] +' , how are you doing?'

    # updated state returned
    return state

# ---Step 3: create graph schema
graph=StateGraph(AgentState)

# add the node to our graph
graph.add_node("greeter", greet)

# Set starting point
# Remember: To connect the nodes together we use edges
graph.add_edge(START, "greeter") # the START is where execution of the graph begins

# Set the endpoint
graph.add_edge("greeter", END)

# compile our graph to an executable
app=graph.compile()

#---Step 4: invoke to run the graph
# test our graph using the invoke keyword
result=app.invoke({"message":"Allan"})

# view the current updated graph state
print(f"Current Graph State:\n{result['message']}")


"""
===========Ignore this part: Is my result on CMD=======
cmd>> python greetings.py
Current Graph State:
Hey Allan , how are you doing?

"""

Por qué es importante el estado

Imagínese un formulario de aplicación con varias páginas: las páginas posteriores siguen viendo las respuestas anteriores. El estado de LangGraph es ese conjunto compartido; cada nodo lo modifica y las conexiones transmiten la actualización hacia adelante.

Name: Allan
Age: 25
Country: Kenya
Course: Information Technology

Conexiones condicionales

Los agentes reales no siguen un único camino. Una función de enrutador examina el estado y elige el siguiente nodo; por ejemplo, entre área y perímetro en una calculadora rectangular:

#!/usr/bin/python3
from typing import TypedDict
from langgraph.graph import StateGraph

# State Schema: Shared amoung all nodes
class AgentState(TypedDict):
    length: int
    width: int
    area: int
    perimeter: int
    operation: str     # we will use it in our router function

# Node--> Function that performs an action
def get_area(state: AgentState)-> AgentState:
    """Calculate the area"""
    state['area']=state['length'] * state ['width']

    return state

# Another Node for calculating perimeter
def get_perimeter(state: AgentState)-> AgentState:
    """Calculate the Perimeter"""
    state['perimeter']=2*(state['length'] + state['width'])

    return state

# Router---> Decides what function to run. Either get_area() or get_perimeter().
def decide_next_node(state: AgentState)-> AgentState:
    """Decide the next node to run"""
    # u will see this when we invoke the graph changing the operation value in state.
    if state['operation']=='a':
        return "area_operation"
    elif state['operation']=='p':
        return "perimeter_operation"

# Graph Schema
graph=StateGraph(AgentState)

# add our two nodes
graph.add_node('area', get_area)
graph.add_node('perimeter', get_perimeter)
graph.add_node('router', lambda state:state) # pass through function. Initial state passes through here

# Below is a quicker way to set the start of the graph without using START being imported
graph.set_entry_point('router')

# Now add conditional edge--> Selects the node to run using our router(decide_next_node)
graph.add_conditional_edges(
    'router', #the node we added
    decide_next_node, # the function to run inorder to decide what to execute next
    {
        # the function decide_next_node returns strings that we match with the names given to our nodes
        "area_operation":"area",
        "perimeter_operation":"perimeter"
    }
)

# Connect both our nodes to the END
# Only one node will be run by our conditional edge. Either we get perimeter or area
# this is another way to connect nodes to the END without importing END
graph.set_finish_point("area")
graph.set_finish_point("perimeter")

# Compile--> make our graph executable
app=graph.compile()

print("---------------We will demonstrate Area fisrt--------")
result=app.invoke({'length':15, 'width':6, 'operation':'a'})

print(f"The Area is: {result['area']}")
print('\n---------------------------------------------------')

"""
=======Dont Add this part: this is my cmd result==============
area_perimeter>python area_perimeter_rect.py
---------------We will demonstrate Area fisrt--------
The Area is: 90

---------------------------------------------------
"""


print("---------------We will demonstrate Perimeter--------")
result=app.invoke({'length':20, 'width':10, 'operation':'p'})

print(f"The Perimeter is: {result['perimeter']}")
print('\n---------------------------------------------------')

"""
======Dont Add this part: this is my cmd result========
area_perimeter>python area_perimeter_rect.py
---------------We will demonstrate Area fisrt--------
The Area is: 90

---------------------------------------------------
---------------We will demonstrate Perimeter--------
The Perimeter is: 60

---------------------------------------------------

"""

Comprobadores y herramientas

Un comprobador toma instantáneas del estado para poder recordar las acciones anteriores más tarde (en memoria para demostraciones; SQLite o Postgres en entornos de producción). Las herramientas son funciones que el modelo puede llamar: calculadora, información meteorológica, búsqueda. Regístrelas con documentación clara; ToolNode ejecuta lo que el modelo solicite en lugar de tener que escribir manualmente cada nodo de herramienta.

from langchain_core.tools import tool

@tool
def current time():
  """Return the current local time"""
  return datetime.now().strftime("%H:%M")

tools=[
  current_time,
]

# Provide tools to LLM
llm=ChatOpenAI(model="gpt-4.1")

llm_with_tools=llm.bind_tools(tools)

Un agente que combina herramientas y memoria:

#!/usr/bin/python3
from datetime import datetime
from dotenv import load_dotenv
import os
from langchain_groq import ChatGroq
#from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
from langchain_core.messages import HumanMessage
from langgraph.graph import StateGraph
from langgraph.graph import add_messages
from langgraph.prebuilt import ToolNode
from typing import Annotated
from typing_extensions import TypedDict
# import a helper conditional routing
from langgraph.prebuilt import tools_condition
# add memory using
from langgraph.checkpoint.memory import MemorySaver

# load our .env file to this file
load_dotenv()

# Our state schema
class AgentState(TypedDict):
    # messages returned by the individual nodes are appended here
    messages: Annotated[list, add_messages]

# add the tools.
# '@' decorator turns our function to a tool
@tool
def current_time():
    """Return the current local time."""
    return datetime.now().strftime("%H:%M:%S")

# add our tools to a list in order to bind them to the llm
tools=[
    current_time, # we have one tool
]

# set up our Groq LLM
#- remember the groq API we used in building vision.py in Part 1. We can use  that
#- Add the API key to a .env file like we did

llm=ChatGroq(
    temperature=0.3,
    groq_api_key=os.getenv("GROQ_API_KEY"),
    model_name="openai/gpt-oss-120b"   # our model in here we pick this.
)
"""
llm = ChatOpenAI(
    model="gpt-5-nano",
    openai_api_key=os.getenv("GPT_5")
)
"""

# Bind our tools to the model
llm_with_tools=llm.bind_tools(tools)


# Add a Node
def chatbot(state:AgentState)-> AgentState:
    """Send in messages to our LLM"""
    # call the LLM
    response=llm_with_tools.invoke(state["messages"])

    # append our LLM response to our state
    return {
        "messages":[response]
    }

# Buiild the graph blueprint
graph=StateGraph(AgentState)

# add nodes
graph.add_node("chatbot", chatbot)
graph.add_node("tools", ToolNode(tools))  # we covered this

# Add edges--> Tells the graph what to execute next.
graph.set_entry_point('chatbot')

# Decides if the LLM needs a tool
graph.add_conditional_edges(
    'chatbot',
    tools_condition, # we imported this
)

# Bring back execution to the LLM after running chatbot
graph.add_edge('tools', 'chatbot')
graph.set_finish_point('chatbot')

# ---NOW add memory to our graph
memory=MemorySaver()

# turn graph to an executable
app=graph.compile(
    checkpointer=memory,
)

# After adding memory we need to set up the threat_id
# Each conversation has a different thread_id and LangGraph will
# call each one with its unique id
config={
    "configurable":{
        "thread_id":"user1"
    }
}

# lets test if our agent responds
while True:
    user_input=input("\nYOU: ")

    # check if exit or quit and terminate program
    if user_input =='exit' or user_input == 'quit':
        print("\nGoodbye!")
        break

    elif user_input=='':
        # if user input is empty restart the infinity loop
        continue

    # invoke our graph
    response=app.invoke(
        {
            "messages":[
                HumanMessage(content=user_input)
            ]
        },
        config=config # add our checkpointer here
    )

    # display output
    print(f"AI🤖: {response['messages'][-1].content}") # retrieve the last message in our dictionary


"""
============Do not Add: This is my CMD results===================
agent_mem_tool>python agent_memory_tool.py

YOU: Hey, I am Fred a CS student. I love skating on Rollerblades.
AI🤖: Hey Fred! 👋 Great to meet a fellow CS enthusiast who also loves skating on Rollerblades.

**CS side:** What areas are you diving into right now—algorithms, AI, systems, graphics, something else? If you ever want project ideas that blend coding with skating (e.g., a skate‑track visualizer, a sensor‑based trick analyzer, or a simple game), just let me know!

YOU:

YOU: Can You tell me the time right now and date
AI🤖: It’s currently **00:11:01** on **2026‑08‑28** (local time).


YOU: me in one statement
AI🤖: Fred is a computer‑science student who fuels his creativity and focus by roller‑blading whenever he can.

"""

Proyecto final: asistente de voz basado en agentes

Reutilice la clave Groq de la primera parte. Prefiera un modelo de texto potente para el bucle del agente; mantenga el SKU de visión para las herramientas relacionadas con imágenes. Distribución:

cmd>> dir
tools main.py mcp_server.py stt.py tts.py .env .gitignore

Módulo de conversión de voz a texto

stt.py puede llamar a Whisper alojado en Groq cuando el hardware local es lento, o ejecutar Whisper localmente cuando hay una GPU disponible.

#!/usr/bin/python3
import os
import speech_recognition as sr
from groq import Groq
from dotenv import load_dotenv

load_dotenv()
client = Groq(api_key=os.environ.get("GROQ_API_KEY"))

# Audio to Groq's whisper model for transcription
def listen():
    # create the speech recognizer object
    recognizer = sr.Recognizer()
    # open the default microphone
    with sr.Microphone() as source:
        #Notify the user recording has started.
        print("\n🎤Listening...")
        # Listen until the user stops speaking
        # recorded audio is stored in a variable 'audio'
        audio = recognizer.listen(source)

    try:
        # speech_recognition gives us the captured audio as WAV bytes,
        # which we can send straight to Groq's transcription endpoint
        wav_bytes = audio.get_wav_data()
        # send the recorded audio to Groq's Whisper model
        # and convert the spoken words into text
        result = client.audio.transcriptions.create(
            file=("mic_input.wav", wav_bytes),
            model="whisper-large-v3",
            response_format="text",
            language="en",
        )
        text = result.strip()
        # return the text in lowercase
        return text.lower()

    # Exception when nothing usable was returned (empty/failed transcription)
    except Exception as e:
        print(f"Transcription error: {e}")
        # Return empty string, not the error text itself - the caller
        return ""

Módulo de texto a voz

tts.py puede hacer funcionar Piper localmente. Instale, descargue una voz y luego sintetice el audio:

cmd>> uv pip install piper-tts

cmd>> mkdir piper_voices

cmd>> cd piper_voices

# Get the voice from hugging face with corresponding json
cmd>> curl -L -o en_US-amy-medium.onnx https://huggingface.co/rhasspy/piper-voices/resolve/main/en/en_US/amy/medium/en_US-amy-medium.onnx

cmd>> curl -L -o en_US-amy-medium.onnx.json https://huggingface.co/rhasspy/piper-voices/resolve/main/en/en_US/amy/medium/en_US-amy-medium.onnx.json

cmd>> cd ..

# when you get back to root directory you should have
cmd>> dir
tools main.py mcp_server.py stt.py tts.py .env .gitignore piper_voices

#!/usr/bin/python3
from piper import PiperVoice
import sounddevice as sd
import numpy as np

MODEL_PATH="piper_voices/en_US-amy-medium.onnx"

# Load once at startup — reuse across calls, loading the model is the slow part
voice = PiperVoice.load(MODEL_PATH)

def speak(text: str):
    """Synthesize text to speech and play it immediately."""
    audio_chunks = []
    for chunk in voice.synthesize(text):
        audio_chunks.append(np.frombuffer(chunk.audio_int16_bytes, dtype=np.int16))

    audio = np.concatenate(audio_chunks)
    sd.play(audio, samplerate=voice.config.sample_rate)
    sd.wait()  # block until playback finishes

Orquestación en main.py

Cargue el servidor MCP de la Primera Parte como herramientas, importe los auxiliares STT/TTS, compile un agente LangGraph con un checkpointer (cámbielo a SQLite para mayor durabilidad) y ejecute el ciclo: escuchar → razonar/herramienta → hablar.

#!/usr/bin/python3
import asyncio
from typing import Annotated, TypedDict
from langchain.chat_models import init_chat_model
from langchain_core.messages import AnyMessage, HumanMessage, SystemMessage
from langgraph.graph import StateGraph, END
from langgraph.graph.message import add_messages
from langgraph.prebuilt import ToolNode, tools_condition
from langchain_mcp_adapters.client import MultiServerMCPClient
from langgraph.checkpoint.memory import MemorySaver
from langchain_groq import ChatGroq
# from langchain_openai import ChatOpenAI
from dotenv import load_dotenv
import os

# This are our user defined libraries from tts.py and stt.py
from stt import listen
from tts import speak


# our env variables loaded here
load_dotenv()

# Set up our LLM-> If you have GPT API comment this out
llm=ChatGroq(
    temperature=0.3,
    groq_api_key=os.getenv("GROQ_API_KEY"),
    model_name="openai/gpt-oss-120b",
    reasoning_effort="low"
)

# Comment out GPT if you dont have the paid API and use the free Groq.
"""
llm = ChatOpenAI(
    model="gpt-5-nano",
    openai_api_key=os.getenv("GPT_5")
)
"""

# We need this of Groq to prevent the model from hallucinating. GPT will work fine with a smaller prompt
SYSTEM_PROMPT = """
You are Called Botnet, a concise voice assistant. Your replies are spoken aloud, not read.

Tools: pick the single right tool, follow its schema exactly, never invent arguments, never repeat a successful call, never fabricate results. If a tool fails and can't be fixed, say so.

Speaking style: talk like a person, not a document. No emojis, no markdown, no symbols like * # - _ / \\ | ~ < > or brackets. Say numbers and dates the way you'd speak them. Use only periods, commas, and question marks. Answer only what was asked, in 1 to 3 sentences unless more detail is truly needed.
"""


# -----Our State: 'add_messages' is the reducer that appends new messages rather
# than overwritingt the list on each update.
class AgentState(TypedDict):
    messages: Annotated[list[AnyMessage], add_messages]


# Remember Part 1 communicaion to the MCP server is asynchronous communication
async def main():
    # 1. Configure MCP server connections
    # We are using stdio because it is running locally
    client=MultiServerMCPClient(
        {
            "tools":{
                "transport":"stdio",
                "command":"python",
                "args":["mcp_server.py"],
            },
            # If we had a remote one:
            # "wikipedia":{
            #       "transport":"streamable_http",
            #       "url":"https://<domain/ServerIP>/mcp",
            #},
        }
    )
    # 2. Discover tools exposed by the connected MCP server
    # Each MCP tools is automatically wrapped as a LangChain BaseTool,
    # with name/description/schema taken from the server's tool spec.
    tools=await client.get_tools()
    tools_by_name={t.name: t for t in tools}

    #display on screen the loaded tools
    print(f"Loaded {len(tools)} tools:{[t.name for t in tools]}\n")
    speak("Welcome Sir!!")

    # Bind up our model
    llm_with_tools=llm.bind_tools(tools)

    # Node--> a python function to carry one task.
    # Get user Speech Transcribed
    def get_input(state:AgentState)-> AgentState:
        """User inputs his command via microphone and they are transcribed to text"""
        text=listen()
        print(f"YOU Said: {text}")

        # the add_reducer function returns only the new messsage
        return {"messages":[HumanMessage(content=text)]}

    # Takes the last message now, the user input and executed using the LLM+MCP
    async def call_model(state: AgentState):
        """Call the LLM passing the user question"""
        messages=[SystemMessage(content=SYSTEM_PROMPT)]+state["messages"]
        response=await llm_with_tools.ainvoke(messages)
        # append our results to our state--> we are just adding new messages remember
        return {"messages":[response]}

    # The speaking node to take the last AI message and speak it
    def talk_back(state: AgentState):
        """Return the message as output voice"""
        text=state["messages"][-1].content
        print(f"AI: {text}")
        speak(text)
        #Node functions should return a state update even when empty to prevent error
        return {}

    # add our MCP server as a tool Node
    tool_node=ToolNode(tools)

    # Build our graph schema
    graph=StateGraph(AgentState)

    # Add our nodes
    graph.add_node("llm", call_model)
    graph.add_node("tools", tool_node)
    graph.add_node("listening", get_input)
    graph.add_node("speaking", talk_back)

    # add the edges-->control the next node to execute in the graph
    graph.set_entry_point("listening")
    graph.add_edge("listening", "llm")

    # here we have a conditional edge when our LLM needs a tool
    # tools_condition we imported will return "tools" if LLM asked for tools
    # or END otherwise.  Mapping END to speaking
    graph.add_conditional_edges(
        "llm",
        tools_condition,
        {"tools":"tools", END:"speaking"}
    )

    # after tools run we loop back to LLM with the  result
    graph.add_edge("tools", "llm")

    # end execution
    graph.set_finish_point("speaking")

    # The memory we will use
    memory=MemorySaver()

    # compile the graph
    app=graph.compile(
        checkpointer=memory
    )

    # set the the thread id since we are using checkpointer memory
    config={
        "configurable":{
            "thread_id":"user1"
        }
    }

    while True:
        await app.ainvoke({"messages":[]}, config=config)


if __name__=="__main__":
    asyncio.run(main())

"""
===========Do Not add this Part: It is my CMD Output(Using Groq Free API)===========
final_Project>python main.py
Loaded 7 tools:['current_time', 'calculate', 'image_info', 'internet_search', 'scrape_webpage', 'weather_forecast', 'system_info']


🎤Listening...
YOU Said: hello, i am Fred.
AI: Hello Fred. Nice to meet you. How can I help you today?

🎤Listening...
YOU Said: introduce yourself in three words.
AI: Helpful. Concise. Friendly.

🎤Listening...
YOU Said: i need to check the weather of a particular city.
AI: Sure, which city would you like the forecast for?

🎤Listening...
YOU Said: nairobi.
AI: The current weather in Nairobi is about twenty three degrees Celsius with broken clouds and moderate humidity.Let me know if you need a longer forecast or anything else.

🎤Listening...
YOU Said: okay, thanks.
AI: You’re welcome. Anything else I can do for you?

"""

Ideas para mejorar la seguridad: ejecute Whisper, Piper y un modelo Ollama completamente sin conexión a Internet; refuerce el control de límites de uso en las versiones gratuitas; amplíe las herramientas MCP sin cambiar la interfaz de voz.

Conclusión

La ruta de la serie es Herramientas MCP → E/S de voz → Orquestación LangGraph. Toma el proyecto final, ejecútalo localmente e itera. Habrá material adicional sobre LangGraph ampliamente disponible una vez que el esqueleto funcione.

“The real problem is not whether machines think but whether men do”
— John McCarthy (founded the term “Artificial Intelligence”)

Notas de integración para el agente de voz final

Cuando el servidor MCP, la conversión de texto a voz, la síntesis de voz y el bucle LangGraph finalmente se conecten, comienza con pruebas de invocación solo con texto. Introduce cadenas de texto escritas en los mismos nodos a los que más tarde llegará el micrófono. Solo después de que la ruta de las herramientas parezca correcta debería introducirse el audio; de lo contrario, cada fallo parecerá un problema con Whisper.

Los identificadores de hilos y los mecanismos de punto de control requieren una política explícita. Un asistente de voz que olvida la última declaración suena defectuoso; uno que filtra las herramientas del Hilo A al Hilo B es aún peor. Utilice un hilo nuevo por sesión de forma predeterminada, y guarde los datos duraderos en el almacén de perfiles respaldado por MCP en lugar de limitarse a almacenarlos solo en el historial de puntos de control.

Los límites de velocidad en las versiones gratuitas de Groq (o similares) se manifestarán como demostraciones inestables. Coloque las declaraciones hechas por el micrófono en cola, muestre un mensaje hablado como “estoy trabajando en ello”, y evite llamadas al modelo simultáneas causadas por audio entrante no solicitado. Al pasar a un funcionamiento completamente sin conexión con Ollama, Whisper y Piper, vuelva a medir la latencia en la CPU: los usuarios toleran menos pacientemente respuestas lentas cuando están esperando con el micrófono encendido.

Qué significa “terminado” para el proyecto en serie

Declare las verificaciones de aceptación: el agente puede resolver una pregunta aritmética oral mediante la calculadora MCP; puede rechazarla o explicar por qué faltan las claves relacionadas con el clima; la memoria verificada recuerda el nombre del usuario en dos llamadas dentro de un mismo proceso; Piper pronuncia la respuesta final sin cortes. Todo lo que vaya más allá de esto se considera trabajo pendiente. Anote esas verificaciones antes de buscar voces más agradables o modelos más grandes. Una pequeña sección vertical que funcione es mejor que un sistema amplio y desorganizado que solo funcione una vez. Al compartir el proyecto, incluya las entradas de bloqueo uv exactas y los IDs de los modelos para que otros no tengan que depurar intercambios silenciosos de modelos. El valor educativo radica en la cadena completa de procesamiento: MCP, voz y gráfico, no en un componente frágil como una hoja de nieve en una computadora portátil.

Si el bucle de habla parece lento, analiza el perfil antes de cambiar de modelo: el tiempo de STT, la latencia del primer herramienta y el tiempo de TTS suelen superar con creces a la generación de tokens. Corrige la etapa lenta que hayas medido, no aquella en la que sea más divertido hacer ajustes.