FastMCP Voice Stack Часть 2: Оркестрация с использованием LangGraph для речевых агентов.

От MCP и голосового ввода-вывода к циклу агента

В первой части рассматривались MCP, серверы инструментов FastMCP, преобразование речи в текст и обратно, а также демонстрация голосового взаимодействия без использования ИИ. В этой части добавляются LangChain и LangGraph, что позволяет ассистенту рассуждать, вызывать эти инструменты и сохранять краткосрочную память — при этом по возможности используются бесплатные уровни API.

LangChain против LangGraph на одной странице

LangChain соединяет модели (OpenAI, Anthropic, Groq, Ollama и другие) с данными, инструментами и рабочими процессами. LangGraph представляет собой слой оркестрации для агентов, выполняющих длительные операции в несколько шагов и имеющих состояние, представленное в виде графов.

Реалистичный агент должен уметь рассуждать, искать информацию, делать краткие выводы, создавать файлы и отправлять письма. Каждый из этих шагов представлен в виде узла — функции на Python. Узлы — это точки; рёбра — линии между ними.

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

Установите необходимые компоненты в существующую среду из первой части:

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

# if using OpenAI API use langchain_openai instead of langchain_groq

Обычные инструменты LangChain испытывают трудности с долговременным хранением данных, сложной структурой ветвлений, повторными попытками выполнения и паузами для участия человека. LangGraph решает эти проблемы благодаря явному управлению состоянием и связями между элементами.

Лексикон

Граф: весь рабочий процесс. Узел: функция, выполняющая определенную задачу. Ребро: следующий шаг выполнения. Состояние: данные, доступные всем узлам для чтения и обновления. START / END: точки входа и выхода. Метод compile() создает исполняемый граф; метод invoke() запускает его с исходным состоянием.

Граф приветствия

Небольшой пример, подходящий для использования в ноутбуках:

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

"""

Почему важно состояние

Представьте многопáginaную форму: более поздние страницы продолжают видеть ответы с предыдущих этапов. Состояние в LangGraph — это общий набор данных; каждый узел его изменяет, а ребра передают эти обновления дальше.

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

Условные ребра

Реальные агенты не следуют одному и тому же пути. Функция маршрутизатора анализирует состояние и выбирает следующий узел — например, внутреннюю часть или периметр при расчёте прямоугольника:

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

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

"""

Контроллеры состояния и инструменты

Контроллер состояния делает снимки текущего состояния, чтобы позже можно было восстановить предыдущие действия (в демо-версиях — в оперативной памяти; в производственных средах — с использованием SQLite или Postgres). Инструменты — это функции, которые может вызывать модель: калькулятор, прогноз погоды, поиск. Регистрируйте их с чёткими описаниями; ToolNode выполняет то, что запросила модель, вместо того чтобы вручную писать каждый узел инструмента.

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)

Агент, объединяющий инструменты и память:

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

"""

Заключительный проект: голосовой ассистент-агент

Используйте тот же ключ Groq, что и в первой части. Для цикла работы агента предпочтительнее использовать мощную текстовую модель; для инструментов, работающих с изображениями, оставьте используемый SKU. Структура:

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

Модуль преобразования речи в текст

stt.py может обращаться к Whisper, размещенному на Groq, при медленной работе локального оборудования, или запускать Whisper локально при наличии GPU.

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

Модуль текста в речь

tts.py может использовать Piper локально. Установите его, скачайте голос, затем синтезируйте речь:

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

Оркестрация в main.py

Загрузите сервер MCP части первой в качестве инструментов, импортируйте помощники STT/TTS, скомпилируйте агента LangGraph с использованием чекпоинта (для долговечности можно заменить его на SQLite), и создайте цикл: слушать → обрабатывать с помощью инструментов → говорить.

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

"""

Идеи по усилению безопасности: запускать Whisper, Piper и модели Ollama полностью офлайн; ужесточить правила лимитирования скорости работы на бесплатных тарифах; расширять набор инструментов MCP без изменения интерфейса голосового управления.

Итоги

Путь серии — MCP tools → voice I/O → LangGraph orchestration. Возьмите финальный проект, запустите его локально и внесите коррективы. Подробная информация о LangGraph станет широко доступной, как только основа заработает.

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

Примечания по интеграции для финального голосового агента

Когда сервер MCP, сервисы STT и TTS, а также LangGraph наконец будут связаны между собой, начните с тестов вызова только с текстом. Подавайте введенные строки в те же узлы, куда позже будет подаваться информация с микрофона. Только после того, как маршрутизация инструментов будет работать корректно, можно вводить аудио — иначе каждая ошибка будет выглядеть как проблема с Whisper.

Идентификаторы потоков и механизмы сохранения состояния требуют четкой политики управления. Голосовой ассистент, забывающий предыдущие запросы, кажется неисправным; ещё хуже, когда он просачивает инструменты потока A в поток B. По умолчанию следует использовать новый поток для каждой сессии, а стабильные данные следует сохранять в хранилище профилей, поддерживаемом MCP, а не только в истории сохранений состояния.

Ограничения по частоте запросов на бесплатных тарифах Groq (или аналогичных сервисов) приводят к ненадежным демонстрациям работы. Необходимо очередывать голосовые запросы, сообщать «работаю над этим» и избегать одновременных вызовов модели из-за внезапно поступающего аудио. При переходе на полностью офлайн-режим с использованием Ollama, Whisper и Piper необходимо повторно измерить задержку работы на процессоре: пользователи менее терпимо относятся к медленным ответам, когда они ждут с включенной микрофоном.

Как выглядит завершение работы над проектом

Опишите проверки приема: агент может решать устные арифметические задачи с помощью калькулятора MCP; он может отказаться или объяснить причину, если отсутствуют необходимые ключи для работы с погодными данными; запоминаемая мемория восстанавливает имя пользователя при двух вызовах в рамках одного процесса; Piper выговаривает окончательный ответ без обрезок. Всё, что выходит за эти рамки, считается нерешенными задачами. Запишите эти проверки до того, как будете стремиться к более качественному звучанию или большим моделям. Лучше иметь небольшой вертикальный фрагмент, который работает корректно, чем широкий горизонтальный фрагмент, который сработает лишь один раз. При обмене проектом укажите точные значения параметров блокировки uv и идентификаторы моделей, чтобы другие не пытались отладить незаметные замены моделей. Образовательная ценность заключается в собранной инфраструктуре — MCP, голосовой интерфейс, графы — а не в хрупком «снежинке»-ноутбуке.

Если процесс речевого цикла кажется медленным, сначала проанализируйте профиль до замены моделей: время преобразования речи в текст, задержка первого инструмента и время генерации речи часто значительно превышают время создания токенов. Исправляйте ту стадию, которую вы измеряете как медленную, а не ту, в которой проще всего экспериментировать. Сначала выпустите вертикальный фрагмент, а затем будете улучшать его.