Stack vocale FastMCP, partie 2 : Orchestration LangGraph pour des agents parlants

De MCP et d’I/O vocal à un cycle d’agent

La première partie a abordé MCP, les serveurs d’outils FastMCP, la reconnaissance vocale et la synthèse vocale, ainsi qu’une démonstration vocale sans utilisation de modèles LLM. Cette partie ajoute LangChain et LangGraph afin que l’assistant puisse raisonner, appeler ces outils et conserver une mémoire à court terme — toujours en utilisant des niveaux API gratuits lorsque c’est possible.

LangChain contre LangGraph en une page

LangChain relie les modèles (OpenAI, Anthropic, Groq, Ollama et d’autres) aux données, outils et workflows. LangGraph est la couche d’orchestration pour des agents à long terme, multi-étapes et à état, exprimés sous forme de graphes.

Un agent réaliste doit pouvoir raisonner, rechercher, résumer, écrire des fichiers et envoyer des e-mails. Chacune de ces actions correspond à un nœud — une fonction Python. Les nœuds représentent des points ; les arêtes sont les lignes qui les relient.

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

Installez l’ensemble dans l’environnement de la première partie existant :

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

# if using OpenAI API use langchain_openai instead of langchain_groq

Les outils simples de LangChain ont du mal à gérer la mémoire persistante, les branches complexes, les tentatives de réexécution et les pauses nécessitant l’intervention humaine. LangGraph résout ces problèmes grâce à des états et des arêtes explicites.

Vocabulaire

Graphe : l’ensemble du flux de travail. Nœud : une fonction tâche. Arête : ce qui s’exécute ensuite. État : données partagées que chaque nœud peut lire et mettre à jour. START / END : point d’entrée et de sortie. compile() crée un graphe exécutable ; invoke() le lance avec un état initial.

Exemple de graphe de salutation

Un petit exemple adapté aux notebooks :

#!/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?

"""

Pourquoi l’état est important

Imaginez un formulaire en plusieurs pages : les pages ultérieures conservent toujours les réponses précédentes. L’état de LangGraph représente ce sac partagé — chaque nœud le modifie et les arêtes transmettent ces mises à jour.

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

Arêtes conditionnelles

Les agents réels ne suivent pas une seule voie. Une fonction de routage examine l’état et choisit le nœud suivant — par exemple, entre la zone intérieure et le périmètre dans un calculateur de rectangle :

#!/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

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

"""

Checkpointers et outils

Un checkpointer prend des captures d’écran de l’état afin que les tours précédents puissent être rappelés ultérieurement (en mémoire pour les démos ; SQLite ou Postgres en production). Les outils sont des fonctions que le modèle peut appeler — calculateur, météo, recherche. Enregistrez-les avec des docstrings claires ; ToolNode exécute ce que le modèle a demandé au lieu d’écrire manuellement chaque nœud d’outil.

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 agent qui combine des outils et de la mémoire :

#!/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.

"""

Projet final : assistant vocal agent

Réutilisez la clé Groq de la première partie. Préférez un modèle de texte puissant pour le cycle de l’agent ; conservez le SKU de vision pour les outils d’image. Disposition :

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

Module texte-voix

stt.py peut appeler Whisper hébergé sur Groq lorsque le matériel local est lent, ou exécuter Whisper localement lorsqu’une GPU est 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 ""

Module texte-voix

tts.py peut faire fonctionner Piper localement. Installez-le, téléchargez une voix, puis synthétisez :

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

Orchestration dans main.py

Chargez le serveur MCP de la première partie en tant qu’outils, importez les aides STT/TTS, compilez un agent LangGraph avec un checkpointer (utilisez SQLite pour plus de durabilité), et itérez : écouter → raisonner/utiliser un outil → parler.

#!/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?

"""

Idees pour renforcer la sécurité : exécuter Whisper, Piper et un modèle Ollama entièrement hors ligne ; renforcer le contrôle des limites de vitesse sur les abonnements gratuits ; étendre les outils MCP sans modifier l’interface vocale.

Conclusion

Le chemin de la série est MCP tools → voice I/O → LangGraph orchestration. Prenez le projet final, exécutez-le localement et itérez. Des informations plus détaillées sur LangGraph deviendront largement accessibles une fois la structure de base fonctionnelle.

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

Notes d’intégration pour l’agent vocal final

Lorsque le serveur MCP, le STT, le TTS et le cycle LangGraph se rencontreront enfin, commencez par des tests d’appel ne contenant que du texte. Alimentez les chaînes de caractères tapées dans les mêmes nœuds que ceux auxquels le microphone alimentera par la suite les données. Ce n’est qu’après que le routage des outils semble correct que l’audio devrait être pris en compte — sinon, chaque échec ressemblera à un problème avec Whisper.

Les identifiants des threads et les mécanismes de sauvegarde nécessitent une politique explicite. Un assistant vocal qui oublie la dernière phrase prononcée semble défectueux ; un autre qui transfère les outils du Thread A vers le Thread B est encore pire. Utilisez par défaut un thread vierge pour chaque session, et stockez les informations durables dans le magasin de profils pris en charge par MCP plutôt que de les conserver uniquement dans l’historique des sauvegardes.

Les limites de fréquence sur les versions gratuites de Groq (ou équivalents) se traduisent par des démonstrations instables. Enfilez les phrases prononcées au micro dans une file d’attente, affichez un message oral indiquant que vous travaillez dessus, et évitez les appels au modèle simultanés dus à des entrées audio non sollicitées. Lorsque vous passez complètement en mode hors ligne avec Ollama, Whisper et Piper, réévaluez la latence sur le CPU : les utilisateurs supportent moins patiemment des réponses lentes lorsqu’ils attendent avec le micro activé.

À quoi ressemble un projet terminé dans cette série

Déclarez les vérifications d’acceptation : l’agent peut résoudre une question arithmétique orale à l’aide du calculateur MCP ; il peut refuser ou expliquer en cas d’absence de clés météo ; la mémoire vérifiée se souvient du nom de l’utilisateur entre deux appels au sein d’un même processus ; Piper prononce la réponse finale sans coupure. Tout ce qui dépasse cela relève des tâches en attente. Notez ces vérifications avant de chercher des voix plus agréables ou des modèles plus gros. Une petite partie verticale fonctionnelle vaut mieux qu’un ensemble horizontal complexe qui ne fonctionne qu’une seule fois. Lorsque vous partagez le projet, incluez les entrées de verrouillage uv exactes ainsi que les identifiants des modèles afin que d’autres ne tentent pas de déboguer des remplacements de modèles silencieux. La valeur éducative réside dans le pipeline assemblé — MCP, voix, graphes — et non dans un élément fragile comme une flocon de neige sur un ordinateur portable.

Si le cycle de parole semble lent, analysez les performances avant de changer de modèle : le temps de reconnaissance vocale en texte, la latence du premier outil et le temps de synthèse vocale en texte prennent souvent bien plus de temps que la génération des tokens. Corrigez l’étape lente que vous mesurez, et non celle qui est la plus facile à ajuster. Déploiez d’abord une version simplifiée, puis itérez.