Multi-agent systems fail in ways single-agent systems don't: infinite loops, contradictory instructions, cascading errors, and agents that silently disagree. This lesson covers the patterns that make agent-to-agent communication reliable.
Select a communication topology to see when to use it, its trade-offs, and sample code.
Linear tasks with clear sequential dependencies. Each agent's output becomes the next agent's input — maximum predictability.
Simple to reason about, easy to debug, deterministic execution order.
No parallelism; one slow agent blocks the entire chain. Cannot handle tasks that require backtracking.
from pydantic import BaseModel
from typing import Literal
class AgentHandoff(BaseModel):
from_agent: str
to_agent: str
task_id: str
task_type: Literal["research", "write", "review", "deploy"]
payload: dict
context: list[str]
# Pipeline: researcher → writer → reviewer
def researcher_node(state: MultiAgentState) -> dict:
result = do_research(state["topic"])
handoff = AgentHandoff(
from_agent="researcher",
to_agent="writer",
task_id=state["task_id"],
task_type="write",
payload={"research_brief": result, "target_word_count": 1200},
context=["User requested technical blog post"],
)
return {"handoffs": [handoff], "researcher_output": result}
def writer_node(state: MultiAgentState) -> dict:
brief = state["handoffs"][-1].payload["research_brief"]
article = write_article(brief)
handoff = AgentHandoff(
from_agent="writer",
to_agent="reviewer",
task_id=state["task_id"],
task_type="review",
payload={"article": article},
context=state["handoffs"][-1].context,
)
return {"handoffs": state["handoffs"] + [handoff], "draft": article}
# Graph: researcher → writer → reviewer → END
graph.add_edge("researcher", "writer")
graph.add_edge("writer", "reviewer")
graph.add_edge("reviewer", END)Agent A asks Agent B for help. B asks A for clarification. A asks B for confirmation. Infinite loop. Always enforce a maximum iteration budget at the graph level.
from typing import TypedDict
from langgraph.graph import StateGraph, END
class SafeAgentState(TypedDict):
messages: list
iteration_count: int
max_iterations: int
def check_limits(state: SafeAgentState) -> str:
"""Guard: stop if iteration budget exceeded."""
if state["iteration_count"] >= state["max_iterations"]:
return "force_stop"
return "continue"
def agent_node(state: SafeAgentState) -> dict:
response = model.invoke(state["messages"])
return {
"messages": [response],
"iteration_count": state["iteration_count"] + 1,
}
graph = StateGraph(SafeAgentState)
graph.add_node("agent", agent_node)
graph.add_node("force_stop", lambda s: {"messages": [SystemMessage("Max iterations reached.")]})
graph.add_conditional_edges(
"agent",
check_limits,
{"continue": "agent", "force_stop": "force_stop"},
)
graph.add_edge("force_stop", END)
app = graph.compile()
result = app.invoke({
"messages": [HumanMessage("Research this topic...")],
"iteration_count": 0,
"max_iterations": 10, # Hard cap
})# Option A: Fail-fast (stop on first error)
def tool_node_strict(state):
try:
result = execute_tool(state)
return {"tool_result": result}
except Exception as e:
raise # Propagate — halt the entire graph
# Option B: Graceful degradation (log and continue)
def tool_node_graceful(state):
try:
result = execute_tool(state)
return {"tool_result": result, "errors": []}
except Exception as e:
return {
"tool_result": None,
"errors": state.get("errors", []) + [str(e)],
}
# Then: conditional edge checks for errors
def route_after_tool(state):
if state.get("errors"):
return "error_handler"
return "next_node"
# Option C: Retry with exponential backoff
import time
def tool_node_with_retry(state, max_retries=3):
for attempt in range(max_retries):
try:
return {"tool_result": execute_tool(state)}
except Exception as e:
if attempt == max_retries - 1:
raise
time.sleep(2 ** attempt) # 1s, 2s, 4sDesign rule: Set a timeout at every agent boundary. An agent that waits forever for a tool response blocks your entire pipeline. For production agents: 30s for tool calls, 120s for subgraph execution.
In the next lesson, we add a human to the loop — how to pause an agent mid-execution to get human approval before taking irreversible actions.