Implementing Custom Function Tools for Robot Control with Hugging Face Speech-to-Speech

You can extend the Hugging Face Speech-to-Speech pipeline with custom function tools by defining JSON schemas in FunctionTool, registering them via the OpenAI Realtime API, and implementing execution hooks in the server's response handlers.

The Speech-to-Speech repository provides a fully modular, low-latency voice-agent pipeline designed for real-time applications. Implementing custom function tools for robot control allows you to trigger physical actions—such as moving a robotic arm or navigating a mobile base—directly from natural language commands processed by the pipeline.

Understanding the Speech-to-Speech Pipeline Architecture

The system operates as a series of connected processing stages orchestrated by s2s_pipeline.py at src/speech_to_speech/s2s_pipeline.py. Each stage runs in its own thread and communicates through typed queues (AudioInItem, STTOutItem, LMOutItem, AudioOutItem), ensuring minimal latency for real-time actuation.

The four core stages are:

• Voice Activity Detection (VAD) – Detects speech boundaries using VADHandler in VAD/vad_handler.py (default: Silero VAD v5).
• Speech-to-Text (STT) – Transcribes audio via handlers in STT/ (default: Parakeet TDT).
• Language Model (LLM) – Generates responses and tool calls via Chat in LLM/chat.py, supporting OpenAI-compatible APIs or local inference.
• Text-to-Speech (TTS) – Synthesizes output using handlers in TTS/ (default: Qwen3-TTS).

The pipeline exposes an OpenAI Realtime-compatible WebSocket API through api/openai_realtime/server.py, which creates a RealtimeService to route conversation items and manage tool execution.

How Tool Calling Works in the Realtime API

The server implements the OpenAI Realtime protocol, enabling tool definitions via JSON schemas that the LLM can invoke during conversations.

Tool Definition – Tools are represented by the FunctionTool class in LLM/tool_call/function_tool.py. This class renders a Python-style function signature (e.g., def move_robot(x: float, y: float): ...) that gets injected into the system prompt.

Tool Injection – The tool_prompt.py module builds the system prompt containing all registered tool signatures. When the LLM decides to invoke a tool, it outputs the call wrapped in <code>...</code> blocks.

Tool Extraction – The _extract_tools routine in LLM/tool_call/function_call.py parses these blocks and converts them into ResponseFunctionToolCall objects, which are then routed to execution handlers.

Step-by-Step Implementation of Robot Control Tools

Define the Tool Schema

Create a FunctionTool instance describing your robot command. This schema informs the LLM about available actions and their parameters.

from speech_to_speech.LLM.tool_call.function_tool import FunctionTool

robot_move = FunctionTool(
    name="move_robot",
    description="Move the robot to a specified (x, y) coordinate.",
    parameters={
        "type": "object",
        "properties": {
            "x": {"type": "number", "description": "Target X coordinate"},
            "y": {"type": "number", "description": "Target Y coordinate"},
            "speed": {"type": "number", "description": "Movement speed", "default": 1.0},
        },
        "required": ["x", "y"],
    },
)

Register Tools with the Session

During a session.update event, provide the list of tools to the runtime configuration. You can send this from the client or modify the server's RuntimeConfig in api/openai_realtime/runtime_config.py.

# Client-side registration

session_update = {
    "type": "session.update",
    "tools": [robot_move.to_dict()],  # Converts to JSON-compatible dict

}
await client.send_conversation_item(**session_update)

Implement Execution Logic

Hook into the LLMProxy or a custom handler in api/openai_realtime/handlers/response.py to execute the robot command when the server receives a conversation.item.create of type function_call_output.

import json
from speech_to_speech.api.openai_realtime.handlers.response import BaseHandler

class RobotHandler(BaseHandler):
    async def on_function_call_output(self, item):
        args = json.loads(item["output"])
        # Insert your robot SDK call here

        robot.move_to(args["x"], args["y"], speed=args.get("speed", 1.0))
        
        # Send spoken confirmation back to user

        await self.client.send_conversation_item(
            type="assistant",
            content=f"Moved to ({args['x']}, {args['y']}).",
        )

Register RobotHandler in the Realtime service via api/openai_realtime/service.py to activate the execution hook.

Start the Server and Test

Launch the realtime server with your tool-enabled configuration, then connect a client.

export OPENAI_API_KEY=YOUR_KEY
speech-to-speech \
    --stt parakeet-tdt \
    --llm_backend responses-api \
    --tts qwen3 \
    --session_prompt "You control a robot. Use tools when appropriate." \
    --tools "move_robot, pick_object"

Connect the client:

python scripts/listen_and_play_realtime.py --host 127.0.0.1 --port 8765

When you say "Move the robot to (2.5, -1.0)", the LLM emits <code>move_robot(x=2.5, y=-1.0)</code>, the server extracts the call, executes your robot SDK code, and speaks a confirmation without exposing the raw function syntax to the user.

Complete Code Examples

Defining a Pick Object Tool

from speech_to_speech.LLM.tool_call.function_tool import FunctionTool

pick_object = FunctionTool(
    name="pick_object",
    description="Pick up an object identified by its label.",
    parameters={
        "type": "object",
        "properties": {
            "label": {"type": "string", "description": "Name of the object"},
            "grip_strength": {"type": "number", "description": "Strength 0-1", "default": 0.8},
        },
        "required": ["label"],
    },
)

Handling Tool Execution with Error Checking

import json
from speech_to_speech.api.openai_realtime.handlers.response import BaseHandler

class RobotHandler(BaseHandler):
    async def on_function_call_output(self, item):
        try:
            args = json.loads(item["output"])
            label = args["label"]
            grip = args.get("grip_strength", 0.8)
            
            # Execute robot command

            success = robot.pick(label, grip_strength=grip)
            
            response = f"Picked up the {label}." if success else f"Failed to pick up {label}."
        except Exception as e:
            response = f"Error executing command: {str(e)}"
            
        await self.client.send_conversation_item(
            type="assistant",
            content=response,
        )

Full Pipeline Launch Command

speech-to-speech \
    --stt parakeet-tdt \
    --llm_backend responses-api \
    --tts qwen3 \
    --tools "move_robot,pick_object" \
    --session_prompt "You are a robot assistant. Use the available tools to manipulate the environment."

Key Files and Architecture

These files enable the plug-and-play architecture for robot control:

• s2s_pipeline.py – Main entry point that creates processing threads and wires the pipeline queues.
• api/openai_realtime/server.py – WebSocket server implementing the OpenAI Realtime protocol.
• api/openai_realtime/runtime_config.py – Mutable session configuration holding custom tool definitions.
• api/openai_realtime/service.py – Creates RealtimeService and registers execution handlers.
• LLM/tool_call/function_tool.py – Defines the FunctionTool class for schema generation.
• LLM/tool_call/function_call.py – Parses LLM output into structured tool calls.
• api/openai_realtime/handlers/response.py – Receives function_call_output events and forwards to execution logic.
• TTS/qwen3_tts_handler.py – Default TTS for speaking confirmations after robot actions.
• STT/parakeet_tdt_handler.py – Low-latency STT supplying initial user commands.

Summary

• Define tools using FunctionTool in LLM/tool_call/function_tool.py with JSON schemas describing robot commands.
• Register tools via the OpenAI Realtime API session.update events managed in runtime_config.py.
• Implement execution by subclassing BaseHandler in handlers/response.py and overriding on_function_call_output.
• Maintain low latency through the zero-copy queue architecture in s2s_pipeline.py, ensuring real-time robot responsiveness.
• Confirm actions using the TTS stage (e.g., Qwen3-TTS) to provide spoken feedback without interrupting the conversation flow.

Frequently Asked Questions

What is the latency impact of adding custom tools?

The zero-copy queue architecture in s2s_pipeline.py ensures that adding custom tools introduces minimal overhead. Since tool extraction happens in the LLM stage and execution runs in parallel with TTS generation, the primary latency cost is your robot SDK's physical response time, not the pipeline processing.

Can I use ROS with this pipeline?

Yes. The execution hook in api/openai_realtime/handlers/response.py can call any Python SDK, including rospy or rclpy. Simply import your ROS client libraries in the handler class and publish commands to your robot's topic within the on_function_call_output method.

How do I handle tool call failures?

Wrap your robot SDK calls in try-except blocks within the handler's on_function_call_output method. Return error messages as assistant conversation items, which the TTS stage will speak to the user. The pipeline automatically continues listening for new commands after handling the error.

Is local LLM inference supported for tool calling?

Yes. The Chat class in LLM/chat.py supports local inference via Transformers or mlx-lm. When using local backends, ensure your model supports function calling formats, as the _extract_tools routine in function_call.py expects specific XML-like <code> tags or standard JSON formatting depending on your prompt configuration.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}