diff --git a/docs/docs/concepts/multi_agent.md b/docs/docs/concepts/multi_agent.md index d8ef0a73b..3b82b2bd1 100644 --- a/docs/docs/concepts/multi_agent.md +++ b/docs/docs/concepts/multi_agent.md @@ -26,10 +26,131 @@ There are several ways to connect agents in a multi-agent system: - **Hierarchical**: you can define a multi-agent system with [a supervisor of supervisors](https://langchain-ai.github.io/langgraph/tutorials/multi_agent/hierarchical_agent_teams/). This is a generalization of the supervisor architecture and allows for more complex control flows. - **Custom multi-agent workflow**: each agent communicates with only a subset of agents. Parts of the flow are deterministic, and only some agents can decide which other agents to call next. +### Handoffs + +In multi-agent architectures agents can be represented as graph nodes. Each agent node executes its step(s) and decides what to do next — finish execution or route to another agent (including routing to itself, e.g. running in a loop). A common pattern in multi-agent interactions is **handoffs**: one agent handing off control to another agent. Handoffs allow you to specify: + +- __destination__: target agent to navigate to (e.g. name of the node to go to) +- __payload__: [what information to pass](#communication-between-agents) to that agent (e.g. state update) + +To implement handoffs in LangGraph, agent nodes can return [`Command`](./low_level.md#command) object that allows you to combine both control flow and state updates: + +```python +def agent(state) -> Command[Literal["agent", "another_agent"]]: + # the condition for routing/halting can be anything, e.g. LLM tool call / structured output, etc. + goto = get_next_agent(...) # 'agent' / 'another_agent' + return Command( + # Specify which agent to call next + goto=goto, + # Update the graph state + update={"my_state_key": "my_state_value"} + ) +``` + +In a more complex scenario where each agent node is itself a graph (i.e. a [subgraph](./low_level.md#subgraphs)), a node in one of the agent subgraphs might want to navigate to a different agent. For example, if you have two agents `alice` and `bob` (subgraph nodes in a parent graph), and `alice` wants to navigate to `bob`, you can do so by setting `graph=Command.PARENT` in the `Command` object: + +```python +def some_node_inside_alice(state) + return Command( + goto="bob", + update={"my_state_key": "my_state_value"}, + # specify which graph to navigate to (defaults to the current graph) + graph=Command.PARENT, + ) +``` + +!!! note + If you need to support visualization for subgraphs communicating using `Command(graph=Command.PARENT)` you would need to wrap them in a node function with `Command` annotation, e.g. instead of this: + + ```python + builder.add_node(alice) + ``` + + you would need to do this: + + ```python + def call_alice(state) -> Command[Literal["bob"]]: + return alice.invoke(state) + + builder.add_node("alice", call_alice) + ``` + +#### Handoffs as tools + +One of the most common agent types is a ReAct-style tool-calling agents. For those types of agents, a common pattern would be wrapping a handoff in a tool call, e.g.: + +```python +def transfer_to_bob(state): + """Transfer to bob.""" + return Command( + goto="bob", + update={"my_state_key": "my_state_value"}, + graph=Command.PARENT, + ) +``` + +This is a special case of [updating the graph state from tools](../how-tos/update-state-from-tools.ipynb) where in addition the state update, the control flow is included as well. + +!!! important + +If you want to use tools that return `Command`, you can either use prebuilt [`create_react_agent`][langgraph.prebuilt.chat_agent_executor.create_react_agent] / [`ToolNode`][langgraph.prebuilt.tool_node.ToolNode] components, or implement your own tool-executing node that collects `Command` objects returned by the tools and returns a list of them, e.g.: + +```python +def call_tools(state): + ... + commands = [tools_by_name[call["name"].invoke(call, config={"coerce_tool_content": False}) for tool_call in tool_calls] + return commands +``` + +Let's now take a closer look at the different multi-agent architectures. + ### Network In this architecture, agents are defined as graph nodes. Each agent can communicate with every other agent (many-to-many connections) and can decide which agent to call next. This architecture is good for problems that do not have a clear hierarchy of agents or a specific sequence in which agents should be called. + +```python +from typing import Literal +from langchain_openai import ChatOpenAI +from langgraph.graph import StateGraph, MessagesState, START + +model = ChatOpenAI() + +def agent_1(state: MessagesState) -> Command[Literal["agent_2", "agent_3", END]]: + # you can pass relevant parts of the state to the LLM (e.g., state["messages"]) + # to determine which agent to call next. a common pattern is to call the model + # with a structured output (e.g. force it to return an output with a "next_agent" field) + response = model.invoke(...) + # route to one of the agents or exit based on the LLM's decision + # if the LLM returns "__end__", the graph will finish execution + return Command( + goto=response["next_agent"], + update={"messages": [response["content"]]}, + ) + +def agent_2(state: MessagesState) -> Command[Literal["agent_1", "agent_3", END]]: + response = model.invoke(...) + return Command( + goto=response["next_agent"], + update={"messages": [response["content"]]}, + ) + +def agent_3(state: MessagesState) -> Command[Literal["agent_1", "agent_2", END]]: + ... + return Command( + goto=response["next_agent"], + update={"messages": [response["content"]]}, + ) + +builder = StateGraph(MessagesState) +builder.add_node(agent_1) +builder.add_node(agent_2) +builder.add_node(agent_3) + +builder.add_edge(START, "agent_1") +network = builder.compile() +``` + ### Supervisor In this architecture, we define agents as nodes and add a supervisor node (LLM) that decides which agent nodes should be called next. We use [conditional edges](./low_level.md#conditional-edges) to route execution to the appropriate agent node based on supervisor's decision. This architecture also lends itself well to running multiple agents in parallel or using [map-reduce](../how-tos/map-reduce.ipynb) pattern. @@ -37,43 +158,41 @@ In this architecture, we define agents as nodes and add a supervisor node (LLM) ```python from typing import Literal from langchain_openai import ChatOpenAI -from langgraph.graph import StateGraph, MessagesState, START +from langgraph.graph import StateGraph, MessagesState, START, END model = ChatOpenAI() -class AgentState(MessagesState): - next: Literal["agent_1", "agent_2", "__end__"] - -def supervisor(state: AgentState): +def supervisor(state: MessagesState) -> Command[Literal["agent_1", "agent_2", END]]: # you can pass relevant parts of the state to the LLM (e.g., state["messages"]) # to determine which agent to call next. a common pattern is to call the model # with a structured output (e.g. force it to return an output with a "next_agent" field) response = model.invoke(...) - # the "next" key will be used by the conditional edges to route execution - # to the appropriate agent - return {"next": response["next_agent"]} + # route to one of the agents or exit based on the supervisor's decision + # if the supervisor returns "__end__", the graph will finish execution + return Command(goto=response["next_agent"]) -def agent_1(state: AgentState): +def agent_1(state: MessagesState) -> Command[Literal["supervisor"]]: # you can pass relevant parts of the state to the LLM (e.g., state["messages"]) # and add any additional logic (different models, custom prompts, structured output, etc.) response = model.invoke(...) - return {"messages": [response]} + return Command( + goto="supervisor", + update={"messages": [response]}, + ) -def agent_2(state: AgentState): +def agent_2(state: MessagesState) -> Command[Literal["supervisor"]]: response = model.invoke(...) - return {"messages": [response]} + return Command( + goto="supervisor", + update={"messages": [response]}, + ) -builder = StateGraph(AgentState) +builder = StateGraph(MessagesState) builder.add_node(supervisor) builder.add_node(agent_1) builder.add_node(agent_2) builder.add_edge(START, "supervisor") -# route to one of the agents or exit based on the supervisor's decisiion -# if the supervisor returns "__end__", the graph will finish execution -builder.add_conditional_edges("supervisor", lambda state: state["next"]) -builder.add_edge("agent_1", "supervisor") -builder.add_edge("agent_2", "supervisor") supervisor = builder.compile() ``` @@ -121,37 +240,29 @@ To address this, you can design your system _hierarchically_. For example, you c ```python from typing import Literal from langchain_openai import ChatOpenAI -from langgraph.graph import StateGraph, MessagesState, START +from langgraph.graph import StateGraph, MessagesState, START, END model = ChatOpenAI() # define team 1 (same as the single supervisor example above) -class Team1State(MessagesState): - next: Literal["team_1_agent_1", "team_1_agent_2", "__end__"] -def team_1_supervisor(state: Team1State): +def team_1_supervisor(state: MessagesState) -> Command[Literal["team_1_agent_1", "team_1_agent_2", END]]: response = model.invoke(...) - return {"next": response["next_agent"]} + return Command(goto=response["next_agent"]) -def team_1_agent_1(state: Team1State): +def team_1_agent_1(state: MessagesState) -> Command[Literal["team_1_supervisor"]]: response = model.invoke(...) - return {"messages": [response]} + return Command(goto="team_1_supervisor", update={"messages": [response]}) -def team_1_agent_2(state: Team1State): +def team_1_agent_2(state: MessagesState) -> Command[Literal["team_1_supervisor"]]: response = model.invoke(...) - return {"messages": [response]} + return Command(goto="team_1_supervisor", update={"messages": [response]}) team_1_builder = StateGraph(Team1State) team_1_builder.add_node(team_1_supervisor) team_1_builder.add_node(team_1_agent_1) team_1_builder.add_node(team_1_agent_2) team_1_builder.add_edge(START, "team_1_supervisor") -# route to one of the agents or exit based on the supervisor's decisiion -# if the supervisor returns "__end__", the graph will finish execution -team_1_builder.add_conditional_edges("team_1_supervisor", lambda state: state["next"]) -team_1_builder.add_edge("team_1_agent_1", "team_1_supervisor") -team_1_builder.add_edge("team_1_agent_2", "team_1_supervisor") - team_1_graph = team_1_builder.compile() # define team 2 (same as the single supervisor example above) @@ -174,31 +285,22 @@ team_2_graph = team_2_builder.compile() # define top-level supervisor -class TopLevelState(MessagesState): - next: Literal["team_1", "team_2", "__end__"] - -builder = StateGraph(TopLevelState) -def top_level_supervisor(state: TopLevelState): +builder = StateGraph(MessagesState) +def top_level_supervisor(state: MessagesState): # you can pass relevant parts of the state to the LLM (e.g., state["messages"]) # to determine which team to call next. a common pattern is to call the model # with a structured output (e.g. force it to return an output with a "next_team" field) response = model.invoke(...) - # the "next" key will be used by the conditional edges to route execution - # to the appropriate team - return {"next": response["next_team"]} + # route to one of the teams or exit based on the supervisor's decision + # if the supervisor returns "__end__", the graph will finish execution + return Command(goto=response["next_team"]) -builder = StateGraph(TopLevelState) +builder = StateGraph(MessagesState) builder.add_node(top_level_supervisor) builder.add_node(team_1_graph) builder.add_node(team_2_graph) builder.add_edge(START, "top_level_supervisor") -# route to one of the teams or exit based on the supervisor's decision -# if the top-level supervisor returns "__end__", the graph will finish execution -builder.add_conditional_edges("top_level_supervisor", lambda state: state["next"]) -builder.add_edge("team_1_graph", "top_level_supervisor") -builder.add_edge("team_2_graph", "top_level_supervisor") - graph = builder.compile() ``` @@ -208,7 +310,7 @@ In this architecture we add individual agents as graph nodes and define the orde - **Explicit control flow (normal edges)**: LangGraph allows you to explicitly define the control flow of your application (i.e. the sequence of how agents communicate) explicitly, via [normal graph edges](./low_level.md#normal-edges). This is the most deterministic variant of this architecture above — we always know which agent will be called next ahead of time. -- **Dynamic control flow (conditional edges)**: in LangGraph you can allow LLMs to decide parts of your application control flow. This can be achieved by using [conditional edges](./low_level.md#conditional-edges). A special case of this is a [supervisor tool-calling](#supervisor-tool-calling) architecture. In that case, the tool-calling LLM powering the supervisor agent will make decisions about the order in which the tools (agents) are being called. +- **Dynamic control flow (Command)**: in LangGraph you can allow LLMs to decide parts of your application control flow. This can be achieved by using [`Command`](./low_level.md#command). A special case of this is a [supervisor tool-calling](#supervisor-tool-calling) architecture. In that case, the tool-calling LLM powering the supervisor agent will make decisions about the order in which the tools (agents) are being called. ```python from langchain_openai import ChatOpenAI