Command and Send

Learn how Galileo logs LangGraph Command and Send types for advanced control flow

Overview

LangGraph provides two types for advanced graph control flow:

• Command — Combines a state update with routing in a single return value. Instead of using conditional edges, a node can return a Command with an update and goto to update state and control which node runs next.
• Send — Enables map-reduce (fan-out) patterns by dispatching work to a node with custom state. Returning a list of Send objects from an edge function runs the target node once per Send, in parallel.

Both types are useful for building dynamic, multi-path agents in LangGraph.

Galileo support

Galileo’s GalileoCallback and GalileoAsyncCallback handle Command and Send automatically. No additional configuration is needed — just pass the callback as usual and all control flow data is captured.

Using Command

Command lets a node update state and route to the next node in one step, replacing the need for separate conditional edges.

Python

from typing import TypedDict

from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, START, END
from langgraph.types import Command

from galileo.handlers.langchain.handler import GalileoCallback


class State(TypedDict):
    query: str
    category: str
    response: str


def classify(state: State) -> Command:
    """Classify the query and route to the appropriate handler."""
    query = state["query"]
    # In practice, you might use an LLM to classify the query
    if "refund" in query.lower():
        return Command(update={"category": "billing"}, goto="billing_handler")
    return Command(update={"category": "general"}, goto="general_handler")


def billing_handler(state: State) -> dict:
    return {"response": f"Billing support for: {state['query']}"}


def general_handler(state: State) -> dict:
    return {"response": f"General support for: {state['query']}"}


# Build the graph
graph = StateGraph(State)
graph.add_node("classify", classify)
graph.add_node("billing_handler", billing_handler)
graph.add_node("general_handler", general_handler)
graph.add_edge(START, "classify")
graph.add_edge("billing_handler", END)
graph.add_edge("general_handler", END)

app = graph.compile()

# Run with Galileo logging — Command routing is captured automatically
galileo_callback = GalileoCallback(project="langgraph-command-example")
result = app.invoke(
    {"query": "I need a refund for my last order"},
    config={"callbacks": [galileo_callback]},
)
print(result["response"])

import { GalileoCallback, GalileoLogger } from "galileo";
import {
  Annotation,
  StateGraph,
  START,
  END,
  Command,
} from "@langchain/langgraph";

// Define the state
const StateAnnotation = Annotation.Root({
  query: Annotation<string>,
  category: Annotation<string>,
  response: Annotation<string>,
});

// Classify the query and route to the appropriate handler
function classify(state: typeof StateAnnotation.State): Command {
  const query = state.query;
  // In practice, you might use an LLM to classify the query
  if (query.toLowerCase().includes("refund")) {
    return new Command({
      update: { category: "billing" },
      goto: "billing_handler",
    });
  }
  return new Command({
    update: { category: "general" },
    goto: "general_handler",
  });
}

function billingHandler(
  state: typeof StateAnnotation.State
): Partial<typeof StateAnnotation.State> {
  return { response: `Billing support for: ${state.query}` };
}

function generalHandler(
  state: typeof StateAnnotation.State
): Partial<typeof StateAnnotation.State> {
  return { response: `General support for: ${state.query}` };
}

// Build the graph
const graph = new StateGraph(StateAnnotation)
  .addNode("classify", classify, 
    { ends: ["billing_handler", "general_handler"] })
  .addNode("billing_handler", billingHandler)
  .addNode("general_handler", generalHandler)
  .addEdge(START, "classify")
  .addEdge("billing_handler", END)
  .addEdge("general_handler", END);

const app = graph.compile();

// Run with Galileo logging — Command routing is captured automatically
const logger = new GalileoLogger({
  projectName: "langgraph-command-example",
});
const galileoCallback = new GalileoCallback(logger);

const result = await app.invoke(
  { query: "I need a refund for my last order" },
  { callbacks: [galileoCallback] }
);

console.log(result.response);

In this example, the classify node returns a Command that sets the category field and routes to either billing_handler or general_handler. Galileo captures the full Command return value, including the routing decision.

Using Send

Send enables fan-out patterns where multiple instances of a node run in parallel, each with different input state.

Python

import operator
from typing import Annotated, TypedDict

from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, START, END
from langgraph.types import Send

from galileo.handlers.langchain.handler import GalileoCallback


class OverallState(TypedDict):
    topics: list[str]
    summaries: Annotated[list[str], operator.add]


class SummaryState(TypedDict):
    topic: str
    summaries: Annotated[list[str], operator.add]


def fan_out(state: OverallState) -> list[Send]:
    """Dispatch each topic to a summarize node in parallel."""
    return [Send("summarize", {"topic": t}) for t in state["topics"]]


def summarize(state: SummaryState) -> dict:
    """Summarize a single topic."""
    # In practice, you would call an LLM here
    return {"summaries": [f"Summary of {state['topic']}"]}


# Build the graph
graph = StateGraph(OverallState)
graph.add_node("summarize", summarize)
graph.add_conditional_edges(START, fan_out)
graph.add_edge("summarize", END)

app = graph.compile()

# Run with Galileo logging — each Send is captured automatically
galileo_callback = GalileoCallback(project="langgraph-send-example")
result = app.invoke(
    {"topics": ["AI safety", "Quantum computing", "Climate change"]},
    config={"callbacks": [galileo_callback]},
)
print(result["summaries"])

import { GalileoCallback, GalileoLogger } from "galileo";
import {
  Annotation,
  StateGraph,
  START,
  END,
  Send,
} from "@langchain/langgraph";

// Define the overall state with a reducer that concatenates summary arrays
const OverallStateAnnotation = Annotation.Root({
  topics: Annotation<string[]>,
  summaries: Annotation<string[]>({
    reducer: (current, update) => [...current, ...update],
    default: () => [],
  }),
});

// Define the per-topic state
const SummaryStateAnnotation = Annotation.Root({
  topic: Annotation<string>,
  summaries: Annotation<string[]>({
    reducer: (current, update) => [...current, ...update],
    default: () => [],
  }),
});

// Dispatch each topic to a summarize node in parallel
function fanOut(state: typeof OverallStateAnnotation.State): Send[] {
  return state.topics.map((t) => new Send("summarize", { topic: t }));
}

// Summarize a single topic
function summarize(
  state: typeof SummaryStateAnnotation.State
): Partial<typeof SummaryStateAnnotation.State> {
  // In practice, you would call an LLM here
  return { summaries: [`Summary of ${state.topic}`] };
}

// Build the graph
const graph = new StateGraph(OverallStateAnnotation)
  .addNode("summarize", summarize)
  .addConditionalEdges(START, fanOut)
  .addEdge("summarize", END);

const app = graph.compile();

// Run with Galileo logging — each Send is captured automatically
const logger = new GalileoLogger({
  projectName: "langgraph-send-example",
});
const galileoCallback = new GalileoCallback(logger);

const result = await app.invoke(
  { topics: ["AI safety", "Quantum computing", "Climate change"] },
  { callbacks: [galileoCallback] }
);

console.log(result.summaries);

Here, fan_out returns a list of Send objects — one per topic. LangGraph runs the summarize node for each topic in parallel, and the results are aggregated via a list-concatenation reducer. Galileo logs each parallel execution.

Command with tool messages

When a tool returns a Command, it can include a ToolMessage in the state update. Galileo automatically extracts the ToolMessage for proper tool-call logging.

Python

from typing import Any

from langchain_core.messages import ToolMessage
from langchain_core.tools import tool
from langgraph.types import Command


@tool
def lookup_order(order_id: str) -> Command:
    """Look up an order by ID and return the result as a Command."""
    # Simulate an order lookup
    order_data = {"order_id": order_id, "status": "shipped", "total": 49.99}

    # Return a Command that includes a ToolMessage in the state update.
    # Galileo automatically extracts the ToolMessage for proper logging.
    return Command(
        update={
            "messages": [
                ToolMessage(
                    content=f"Order {order_id}: status={order_data['status']}, total=${order_data['total']}",
                    tool_call_id="placeholder",  # LangGraph fills this automatically
                )
            ]
        }
    )

import { ToolMessage } from "@langchain/core/messages";
import { tool } from "@langchain/core/tools";
import { Command } from "@langchain/langgraph";
import { z } from "zod";

// Define a tool that returns a Command with a ToolMessage
const lookupOrder = tool(
  async ({ orderId }: { orderId: string }): Promise<Command> => {
    // Simulate an order lookup
    const orderData = { orderId, status: "shipped", total: 49.99 };

    // Return a Command that includes a ToolMessage in the state update.
    // Galileo automatically extracts the ToolMessage for proper logging.
    return new Command({
      update: {
        messages: [
          new ToolMessage({
            content: `Order ${orderId}: status=${orderData.status}, total=$${orderData.total}`,
            tool_call_id: "placeholder", // LangGraph fills this automatically
          }),
        ],
      },
    });
  },
  {
    name: "lookup_order",
    description:
      "Look up an order by ID and return the result as a Command.",
    schema: z.object({
      orderId: z.string().describe("The order ID to look up"),
    }),
  }
);

This pattern is useful when a tool needs to both return a result to the LLM (via ToolMessage) and update other parts of the graph state at the same time.

What gets logged

Galileo serializes Command and Send objects and captures their fields in the trace:

Command fields

Field	Description
update	The state update dictionary applied by the command
goto	The target node(s) the command routes to
graph	The subgraph to send the command to (if applicable)
resume	The resume value for interrupt-based workflows

Send fields

Field	Description
node	The target node to dispatch work to
arg	The custom state passed to the target node

All fields are serialized recursively, including nested messages and complex objects.

Next steps

Logging

Set up GalileoCallback for LangChain and LangGraph logging.

Middleware

Use GalileoMiddleware for automatic LangGraph agent logging.

Experiments

Run and track experiments with LangChain.