★ FIELD-TESTED · HOW WE BUILD IT

ROS 2 × LangChain
the agent as a high-level controller

Everything so far was the textbook. This chapter shows how we actually put LLM agents on real robots at the BioRobotics Group (CAR, CSIC-UPM).

ROS 2 JazzyLangChain + LangGraphSafety gateClosed loop

// THE CONCEPT

The LLM never drives the motors. It acts as a high-level controller: it turns human intent into a few bounded commands, a deterministic safety gate validates them, and ROS 2 delivers them to a robot whose real-time control stays classic.

// YOUR ROBOT GAINS

  • Natural-language control of real hardware
  • Hard safety limits the LLM cannot break
  • A slow reasoning loop on top of fast classic control

// KEY TERMS

  • Capability manifest: YAML listing every allowed action + limits
  • Safety gate: code that validates each command
  • High-level controller: decides what, never how at 100 Hz

ROS 2 meets LangChain

If you know ROS 2, you already know half of this. Most agent concepts map directly onto ROS 2 concepts:

ROS 2Agent worldHow we connect them
service / actiontoolEach allowed robot command becomes one LangChain tool
topic subscriptionobservationSensor data summarized into the agent's context
parameter limitstool schemaRanges and max steps written in a manifest, enforced by the gate
nodeagent / graph nodeEach agent runs in (or next to) its own ROS 2 node
launch filegraph compositionController, observer and planner started together, but independent
rosbagtraceEvery prompt, tool call, token count and latency logged for analysis

The architecture

Step through the layers, then press Trace a run to follow one decision around the loop.

Pattern 1 · The safety gate

The most important idea in this chapter: the LLM proposes, the gate decides. Every action the robot can take is declared once in a capability manifest with its range, its maximum change per command and the ROS 2 topic or service it maps to. The same file generates the LLM's tools and the validation code, so they can never drift apart.

Try to make R-80 misbehave:

Rejections are information. A rejected command is returned to the LLM as the tool result ("jump bigger than 0.3 m/s, change it gradually"), so the agent adapts instead of crashing. And the emergency stop never goes through the LLM: it lives in hardware and classic code.

# Capability manifest: the ONLY actions the LLM may take, and their limits.
# Tools AND the safety gate are generated from this one file, the same pattern
# we use to put LLM agents on real robots.
capabilities:
  - id: set_speed
    description: Set R-80's forward speed.
    type: number
    min: 0.0
    max: 0.8
    max_delta: 0.3        # biggest change allowed in a single command
    units: m/s
    initial: 0.0
    topic: /r80/cmd/speed

  - id: set_gripper_force
    description: Set the gripper closing force. Fragile items need a low force.
    type: number
    min: 0
    max: 100
    max_delta: 40
    units: "%"
    initial: 50
    topic: /r80/cmd/gripper_force

  - id: set_mode
    description: Set the driving mode. Use slow inside zone B.
    type: string
    enum: [slow, normal]
    initial: normal
    topic: /r80/cmd/mode
class SafetyGate:
    def __init__(self, manifest: dict) -> None:
        self.caps = {c["id"]: c for c in manifest["capabilities"]}
        self.state = {c["id"]: c["initial"] for c in manifest["capabilities"]}

    def check(self, cap_id: str, value) -> str | None:
        """Return None if the command is safe, or the reason it was rejected."""
        cap = self.caps.get(cap_id)
        if cap is None:
            return f"REJECTED: '{cap_id}' is not in the capability manifest."
        if cap["type"] == "number":
            if not cap["min"] <= value <= cap["max"]:
                return f"REJECTED: {value} is outside [{cap['min']}, {cap['max']}] {cap['units']}."
            if abs(value - self.state[cap_id]) > cap["max_delta"]:
                return (f"REJECTED: {self.state[cap_id]} -> {value} is a jump bigger than "
                        f"{cap['max_delta']} {cap['units']}. Change it gradually.")
        elif value not in cap["enum"]:
            return f"REJECTED: '{value}' is not one of {cap['enum']}."
        return None
def build_tools(gate: SafetyGate, bridge: RosBridge) -> list:
    tools = []
    for cap in MANIFEST["capabilities"]:
        if cap["type"] == "number":
            hint = f"{cap['min']}..{cap['max']} {cap['units']}, change at most {cap['max_delta']} per call"
            field = (float, Field(description=hint))
        else:
            field = (str, Field(description=f"one of {cap['enum']}"))
        Args = create_model(f"{cap['id']}_args", value=field)

        def run(value, _cap=cap) -> str:
            if reason := gate.check(_cap["id"], value):
                WORLD.emit("gate_reject", f"{_cap['id']}({value!r}): {reason}")
                return reason                      # the LLM reads this and can adapt
            bridge.publish(_cap, value)
            gate.state[_cap["id"]] = value
            WORLD.emit("gate_ok", f"{_cap['topic']} <- {value}")
            return f"OK: {_cap['id']} = {value} {_cap.get('units', '')}".strip()

        tools.append(StructuredTool.from_function(
            func=run, name=cap["id"], description=cap["description"], args_schema=Args))

    @tool
    def get_robot_state() -> str:
        """Current commanded values of every capability (closed-loop context)."""
        return str(gate.state)

    return tools + [get_robot_state]

Pattern 2 · Slow loop over fast loop

Slow loop · LLM

Seconds to minutes. Reads windows of metrics, reasons about goals, decides which setting to change. Tolerates latency, benefits from context.

Fast loop · classic control

Milliseconds (100 Hz and up). Motor controllers, impedance, balance. Deterministic, certifiable, never waits for a token.

We also split the reasoning in two: an observer agent that is advisory only (it reads metrics and writes structured observations, but cannot command the robot), and a controller agent that is the only one allowed to call robot tools. Fewer agents with hands means a smaller surface for mistakes.

Pattern 3 · A strict output contract

A robot can't act on "Sure! I think maybe we could…". The controller must answer with structured output. Our pattern: a status, a bounded plan, and a short justification per step so every decision can be audited later.

{
  "status": "EXECUTE",            // or "NEED_MORE_INFO" | "REFUSE"
  "plan": [
    {"primitive": "set_speed", "parameters": {"value": 0.4},
     "justification": "Operator reports a wet floor; reduce speed within the allowed step."}
  ],
  "spoken_response": "Slowing down a little."
}

Swap the manifest, keep the architecture

Nothing in this stack is specific to one robot. The agent, the gate, the bridge and the loops stay the same. What changes is the capability manifest: which commands exist, their limits, and which ROS 2 interface each one maps to.

Mobile robots

Speed, driving mode, waypoints. The gate limits acceleration and forbids zones. The agent plans the route from a spoken goal.

Robot arms

Gripper force, joint velocity limits, named poses. The gate enforces force and workspace limits. The agent picks which pose, not the trajectory.

Multi-robot systems

A LangGraph planner drafts the mission, one controller agent per robot executes through its own gate, and a human approves the plan.

Lessons learned on real robots

  1. Never give the LLM direct access to hardware.Manifest + deterministic gate: ranges, max step per command, rate limits. No path around it.
  2. One manifest, many consumers.Tools, validation, prompts and docs are all generated from the same YAML, so they never disagree.
  3. Keep the LLM in the slow loop.It reasons over seconds of data. Real-time control stays classic.
  4. Separate observers from actors.Advisory agents can analyse anything. Only one agent can command the robot.
  5. Demand structured output.EXECUTE / NEED_MORE_INFO / REFUSE plus a justification per step: parseable and auditable.
  6. Be provider-agnostic.Cloud APIs for development, local or self-hosted models for privacy and cost. One env var switches them (chapter 2).
  7. Log everything.Prompts, tool calls, tokens, latency. You'll need it for debugging, and for your papers.
  8. Always have a deterministic fallback.Timeouts and malformed answers happen. The robot must stay safe when the LLM doesn't answer.

Run it on your ROS 2

The chapter script works with or without ROS. Without ROS 2 it prints the messages it would publish. With ROS 2 sourced it publishes real std_msgs topics (tested on Jazzy).

# terminal 1: no API key needed for the gate demo
source /opt/ros/jazzy/setup.bash
python3 ch7_ros2_bridge.py --no-llm     # drop --no-llm to let the agent drive

# terminal 2: watch what reaches the robot
source /opt/ros/jazzy/setup.bash
ros2 topic echo /r80/cmd/speed

rclpy comes with your ROS 2 installation, not from pip. Run the script with the Python that ROS 2 was built for, with LangChain installed in that environment.

🤖 ASK YOUR AI COPILOT
I have a ROS 2 robot with these services and topics: <list them>. Create a YAML capability manifest with a range and max step per command, generate LangChain tools from it, and add a safety gate that validates every call before publishing to ROS 2. Use code/ch7_ros2_bridge.py as the reference pattern.
Paste into Claude Code, Cursor, Codex… Works best with the LangChain power-ups from the welcome page.

// SAVE POINT

  • The LLM is a high-level controller: it decides what, never how at 100 Hz.
  • Capability manifest + safety gate: the LLM proposes, deterministic code decides.
  • Observer agents advise, one controller acts, humans approve.
  • Same architecture, different robot: swap the manifest.