Module 8: Multi-Agent Orchestration

6. Shared vs Isolated State

Overview

You already know the 4 multi-agent orchestration patterns: Supervisor coordinates centrally, Handoffs transfers control between agents, Subagents delegates with isolated context, Router classifies and directs. Each pattern defines who decides which agent works. But there's a design decision even more fundamental than the orchestration pattern: how the agents communicate with each other. It's the decision that most affects your system's quality, scalability, and debuggability.

Communication between agents boils down to one question: what information does each agent see? At one extreme you have shared state — every agent reads and writes the same state. Simple implementation, trivial coordination, but each agent sees everything the others produce, including information it doesn't need. At the other extreme you have isolated state — each agent has its own private state. Clean context, no interference, but coordinating between agents requires explicit work. Between the two extremes there are variants: message passing, shared memory with namespaces, and the hybrid approach you'll use in the project.

This capsule isn't about implementing one more pattern. It's about making the architectural decision that determines whether your multi-agent system scales or collapses. In production, 80% of quality problems in multi-agent systems don't come from misconfigured agents — they come from badly designed communication.


Shared State

The idea: everyone sees everything

Shared state is the most direct approach. You define a TypedDict with every field any agent needs, and every node in the graph reads and writes that same state:

import operator
from typing import Annotated, TypedDict
from langchain_core.messages import AnyMessage
from langgraph.graph.message import add_messages

class SharedResearchState(TypedDict):
    messages: Annotated[list[AnyMessage], add_messages]
    research_data: str
    analysis_result: str
    report_draft: str
    quality_score: float
    iteration_count: int
    sources_found: list[str]
    current_agent: str

Every agent accesses everything. Look at the researcher — its job is to find information, but it reads fields it doesn't need:

def researcher_node(state: SharedResearchState) -> dict:
    query = state["messages"][-1].content
    previous_analysis = state.get("analysis_result", "")
    sources = state.get("sources_found", [])

    response = model.invoke([
        SystemMessage(content=f"""You are a specialized researcher.
Search for information about: {query}
Previous analysis: {previous_analysis}
Sources already found: {sources}"""),
        *state["messages"],
    ])
    return {"messages": [response], "research_data": response.content}

The analyst does the same — it sees report_draft and iteration_count even though it doesn't need them to analyze data. And the writer sees research_data, quality_score, and the whole message history from the other agents.

Advantages of shared state

Trivial coordination. When the analyst needs the researcher's data, it just reads state["research_data"]. No passing messages, no serialization, no protocols. The data is right there.

Simple debugging. At any point in the flow, you inspect a single object — the state — and you see the whole system.

Fast implementation. One TypedDict, a few nodes, a few edges. It's the first thing you implement when you're prototyping:

from langgraph.graph import StateGraph, START, END

graph = StateGraph(SharedResearchState)
graph.add_node("researcher", researcher_node)
graph.add_node("analyst", analyst_node)
graph.add_node("writer", writer_node)
graph.add_edge(START, "researcher")
graph.add_edge("researcher", "analyst")
graph.add_edge("analyst", "writer")
graph.add_edge("writer", END)

app = graph.compile()

The problem: context bloat

The problem shows up as the system grows. Each iteration piles more data into the state. The messages grow. The intermediate fields fill up. And every agent gets all of it in its prompt:

Iteration 1:
  the researcher sees: the original query (5 tokens of useful context)
  messages: 3 messages

Iteration 3:
  the researcher sees: the query + previous analysis + draft + score + sources
  messages: 15 messages (including analyst and writer outputs it doesn't need)

Iteration 8:
  the researcher sees: the entire accumulated history
  messages: 60+ messages
  90% of the context is noise for a search task

This causes three concrete problems:

  1. Quality degradation. The LLM loses focus with irrelevant information. The researcher starts "responding" to the analyst's analysis instead of finding new information.

  2. Unnecessary cost. Every LLM call includes tokens the agent doesn't need. With 4 agents and 5 iterations, that's 20 calls with the full history.

  3. Interference between agents. One agent reads fields another left in an inconsistent state. The classic bug: the researcher sees report_draft and starts looking for information to "improve the report" instead of researching the original topic.

When to use shared state

Shared state works well in small systems with short flows:

  • Prototypes and MVPs. When you're validating an idea, shared state lets you iterate fast without designing interfaces between agents.
  • Short linear pipelines. If your system is researcher → analyst → writer with no loops, the context doesn't grow uncontrollably.
  • Teams of 2-3 agents. With few agents, the interference is manageable. With 6+, it gets chaotic.
  • Strong dependencies between agents. If the analyst needs exactly what the researcher produces and the writer needs both, shared state avoids duplication.

Isolated State

The idea: each agent in its own bubble

Isolated state inverts the premise: each agent has its own TypedDict with only the fields it needs. It doesn't see others' history, doesn't access their results, doesn't know the global state. You already saw this in the Subagents capsule — now we formalize it as a communication pattern.

class ResearcherState(TypedDict):
    messages: Annotated[list[AnyMessage], add_messages]
    search_query: str
    max_sources: int

class AnalystState(TypedDict):
    messages: Annotated[list[AnyMessage], add_messages]
    raw_data: str
    analysis_type: str

class WriterState(TypedDict):
    messages: Annotated[list[AnyMessage], add_messages]
    key_points: list[str]
    target_length: int
    tone: str

Each agent is an independent subgraph with its own state:

from langgraph.graph import StateGraph, START, END

def build_researcher():
    def search(state: ResearcherState) -> dict:
        response = model.invoke([
            SystemMessage(content=f"Search for information about: {state['search_query']}"),
            *state["messages"],
        ])
        return {"messages": [response]}

    g = StateGraph(ResearcherState)
    g.add_node("search", search)
    g.add_edge(START, "search")
    g.add_edge("search", END)
    return g.compile()

researcher = build_researcher()
analyst = build_analyst()  # the same pattern with AnalystState

The parent as translator

With isolated state, the parent takes charge of translating between agents — it extracts what's relevant from one's result and packages it as the next one's input:

from langchain_core.messages import HumanMessage

class OrchestratorState(TypedDict):
    messages: Annotated[list[AnyMessage], add_messages]
    research_output: str
    analysis_output: str

def delegate_research(state: OrchestratorState) -> dict:
    query = state["messages"][-1].content
    result = researcher.invoke({
        "messages": [HumanMessage(content=query)],
        "search_query": query, "max_sources": 5,
    })
    return {"research_output": result["messages"][-1].content}

def delegate_analysis(state: OrchestratorState) -> dict:
    result = analyst.invoke({
        "messages": [HumanMessage(content="Analyze this data")],
        "raw_data": state["research_output"], "analysis_type": "comparative",
    })
    return {"analysis_output": result["messages"][-1].content}

Each agent receives exactly what it needs, nothing more.

Advantages of isolated state

Constant context. It doesn't matter if the system is 1 iteration in or 50. Each agent receives the same amount of context.

No interference. The researcher can't read report_draft because that field doesn't exist in ResearcherState. Structural enforcement, not discipline.

Independent testing. You can test each agent in isolation with known inputs. If it fails, the bug is isolated.

Scalability. Adding a new agent doesn't affect the existing ones. You define its TypedDict, implement its subgraph, and add the translation logic in the parent.

The cost: explicit coordination

The downside is that all the communication goes through the parent. The parent becomes a "translator" that must know what output each agent produces, what input the next one needs, and how to transform one into the other. If you change the researcher's output format, you have to update the parent's translation logic. With 2 agents that's trivial. With 8, it's a significant point of failure — the parent is coupled to every child's interface.


Message Passing

Beyond shared state and total isolation

Message passing is an intermediate approach where the agents don't share direct state, but communicate through structured messages. Each agent has an "inbox" (what it receives) and an "outbox" (what it produces). The communication is explicit and typed.

The difference from shared state is subtle but important: in shared state, the analyst reads state["research_data"] directly. In message passing, the researcher sends a message with its result, and the analyst receives that message. The researcher decides what to include in the message. The analyst only sees messages addressed to it.

Implementation with typed messages

from dataclasses import dataclass

@dataclass
class AgentMessage:
    sender: str
    receiver: str
    content: str
    message_type: str  # "result", "request", "feedback"

class MessagePassingState(TypedDict):
    messages: Annotated[list[AnyMessage], add_messages]
    agent_inbox: dict[str, list[AgentMessage]]
    current_agent: str

def send_message(state: dict, sender: str, receiver: str,
                 content: str, msg_type: str = "result") -> dict:
    inbox = state.get("agent_inbox", {})
    inbox.setdefault(receiver, []).append(
        AgentMessage(sender=sender, receiver=receiver,
                     content=content, message_type=msg_type))
    return {"agent_inbox": inbox}

def get_inbox(state: dict, agent_name: str) -> list[AgentMessage]:
    return state.get("agent_inbox", {}).get(agent_name, [])

Nodes with an inbox/outbox

Each agent reads its inbox, processes, and sends to the next:

def researcher_with_messages(state: MessagePassingState) -> dict:
    my_inbox = get_inbox(state, "researcher")
    requests = [m.content for m in my_inbox if m.message_type == "request"]
    query = requests[-1] if requests else state["messages"][-1].content

    response = model.invoke([
        SystemMessage(content="You are a researcher. Find relevant information."),
        HumanMessage(content=query),
    ])

    updated = send_message(state, "researcher", "analyst", response.content, "result")
    return {"agent_inbox": updated["agent_inbox"], "messages": [response]}

The analyst does the same: it reads the researcher's messages in its inbox, analyzes, and sends to the writer. Each agent only sees the messages addressed to it — there's no broadcast.

When message passing makes sense

Message passing shines when you need selective communication — the researcher sends to the analyst but not to the writer — or feedback loops where the supervisor asks the researcher for more data without contaminating the writer's inbox.

┌──────────┐  result   ┌──────────┐  result   ┌──────────┐
│Researcher├──────────►│ Analyst  ├──────────►│  Writer  │
└────▲─────┘           └──────────┘           └──────────┘
     │ request
     │
┌────┴─────┐
│Supervisor│
└──────────┘

Shared Memory with the LangGraph Store

Shared memory without context bloat

In Module 6 you learned LangGraph's Store API: InMemoryStore with namespaces to save persistent knowledge across sessions. That same mechanism works for communication between agents — but with an advantage over shared state: each agent accesses only the namespaces that belong to it.

The idea: the agents share a Store with organized namespaces. The researcher writes to its namespace, the analyst reads from the researcher's namespace and writes to its own.

from langgraph.store.memory import InMemoryStore
from langgraph.store.base import BaseStore
from langchain_core.runnables import RunnableConfig

store = InMemoryStore()

Namespaces per agent

The researcher writes to its namespace, the analyst reads from the researcher's namespace:

def researcher_with_store(state: dict, config: RunnableConfig, *, store: BaseStore):
    task_id = config["configurable"].get("task_id", "default")
    response = model.invoke([
        SystemMessage(content="Find relevant information."), *state["messages"],
    ])
    store.put(("tasks", task_id, "researcher"), "results",
              {"data": response.content, "sources": ["source1", "source2"]})
    return {"messages": [response]}

def analyst_with_store(state: dict, config: RunnableConfig, *, store: BaseStore):
    task_id = config["configurable"].get("task_id", "default")
    research = store.get(("tasks", task_id, "researcher"), "results")
    if not research:
        return {"messages": [AIMessage(content="No research data.")]}

    response = model.invoke([
        SystemMessage(content=f"Analyze: {research.value['data']}"),
        HumanMessage(content="Produce a structured analysis."),
    ])
    store.put(("tasks", task_id, "analyst"), "analysis",
              {"result": response.content, "based_on": research.value.get("sources", [])})
    return {"messages": [response]}

The writer would follow the same pattern: it reads from both namespaces (researcher and analyst). Each agent accesses only the namespaces it explicitly looks up — there's no broadcast.

Advantages over pure shared state

  1. Selective access. The researcher can't read what the writer produces. There's no broadcast — each agent explicitly looks up what it needs.
  2. Persistence. The Store survives between invocations. If the system fails mid-run, the partial results are saved.
  3. Rich metadata. Timestamps, versions, quality scores alongside the data. The analyst can decide whether it needs a re-search.
  4. Auditability. store.search(("tasks", task_id)) shows everything each agent produced, in what order, with what metadata.

Compiling with a Store

The graph gets compiled by passing the store, just like you learned in M6:

app = graph.compile(store=store, checkpointer=MemorySaver())
result = app.invoke(
    {"messages": [HumanMessage(content="Research AI agents")]},
    config={"configurable": {"thread_id": "t1", "task_id": "research-001"}},
)

Hybrid Approaches

The recommended pattern

No real system uses pure shared state or pure isolated state. Production systems use a hybrid approach: shared state for high-level coordination, isolated state for execution. The supervisor needs to know who has worked and the quality score (coordination → shared). The researcher doesn't need to see the writer's draft (execution → isolated).

Implementation: coordination state + execution subgraphs

class CoordinationState(TypedDict):
    """Shared state — coordination only, no execution data."""
    messages: Annotated[list[AnyMessage], add_messages]
    current_phase: str
    assigned_agent: str
    iteration_count: int
    quality_score: float
    phase_results: dict[str, str]  # {"researcher": "summary", "analyst": "summary"}

class ResearcherExecState(TypedDict):
    """The researcher's isolated state — only what it needs to search."""
    messages: Annotated[list[AnyMessage], add_messages]
    search_query: str
    found_sources: list[str]

class AnalystExecState(TypedDict):
    """The analyst's isolated state — only data and analysis."""
    messages: Annotated[list[AnyMessage], add_messages]
    raw_data: str
    analysis_framework: str

The supervisor works with CoordinationState. When it delegates to an agent, it extracts what's needed from the coordination state, invokes the agent's subgraph with its own state, and puts the summary back into the coordination state:

researcher_graph = build_researcher()  # a subgraph with ResearcherExecState
analyst_graph = build_analyst()        # a subgraph with AnalystExecState

def supervisor_delegate(state: CoordinationState) -> dict:
    agent = state["assigned_agent"]
    if agent == "researcher":
        query = state["messages"][-1].content
        result = researcher_graph.invoke({
            "messages": [HumanMessage(content=query)],
            "search_query": query, "found_sources": [],
        })
        summary = result["messages"][-1].content[:500]
    elif agent == "analyst":
        research_data = state.get("phase_results", {}).get("researcher", "")
        result = analyst_graph.invoke({
            "messages": [HumanMessage(content="Analyze this data")],
            "raw_data": research_data, "analysis_framework": "comparative",
        })
        summary = result["messages"][-1].content[:500]
    else:
        return {}

    return {
        "phase_results": {**state.get("phase_results", {}), agent: summary},
        "messages": [AIMessage(content=f"[{agent.title()}] {summary}")],
    }

def supervisor_decide(state: CoordinationState) -> dict:
    phases_done = list(state.get("phase_results", {}).keys())
    response = model.invoke([
        SystemMessage(content=f"""Phases completed: {phases_done}.
Quality: {state.get('quality_score', 0)}.
Decide: researcher, analyst, writer, or FINISH."""),
        *state["messages"][-6:],
    ])
    next_agent = response.content.strip().lower()
    if next_agent == "finish":
        return {"current_phase": "done", "assigned_agent": "none"}
    return {"assigned_agent": next_agent, "iteration_count": state.get("iteration_count", 0) + 1}

Why this approach works

The coordination state stays small: the current phase, the assigned agent, counters, and summaries (not complete data) from each agent. The complete data lives in the ResearcherExecState and dies when that subgraph finishes:

┌─────────────────────────────────────────────────────┐
│  COORDINATION STATE (small, coordination only)       │
│  current_phase: "analysis"  │  quality_score: 0.7   │
│  phase_results: {"researcher": "500-char summary"}   │
│                                                      │
│    ┌──────────────┐      ┌──────────────┐            │
│    │  Researcher  │      │   Analyst    │            │
│    │  ExecState   │      │  ExecState   │            │
│    │  (isolated)  │      │  (isolated)  │            │
│    │ search_query │      │ raw_data     │            │
│    └──────────────┘      └──────────────┘            │
└─────────────────────────────────────────────────────┘

Combining with the Store

You can add the Store API to the hybrid approach: supervisor_delegate saves the complete output in the Store (so the next agent can access detailed data) and puts only a summary in phase_results (so the supervisor makes fast decisions). Three complementary layers:

  • Coordination state → summaries for the supervisor's fast decisions
  • Store → complete, persistent data for the agents that need it
  • Execution subgraphs → isolation with minimal context

Comparison

AspectShared StateIsolated StateMessage PassingHybrid
Implementation complexityLowMediumMedium-HighHigh
Context bloatYes, grows with each iterationNoControlled by designNo
Coordination between agentsTrivialRequires the parent as a translatorExplicit, via messagesEasy via the coordination state
DebuggingOne state, easyEach agent isolated, easyTracing messages, mediumState + Store, medium
Scalability (# agents)Poor (>4 agents)GoodGoodGood
Interference between agentsHighNoneLowLow
Data persistenceOnly with a checkpointerLost when the subgraph endsLost when it endsThe Store persists
When to usePrototypes, 2-3 agentsDecomposable tasksComplex feedback loopsProduction

Rule of thumb

  • Prototype / 2-3 linear agents → Shared state
  • 4+ agents / multiple iterations → Hybrid
  • Feedback loops → Message passing or hybrid
  • Production with persistence → Hybrid + Store
  • 100% independent tasks → Isolated state (subagents)

Connection to the Project

This module's Research Agent v5 uses the hybrid approach:

  • The Supervisor's coordination state: current_phase, assigned_agent, iteration_count, quality_score, phase_results (summaries). A small, focused state for routing decisions.

  • Each worker's isolated execution: the Researcher has a ResearcherExecState with search_query and found_sources. The Analyst has an AnalystExecState with raw_data. The Writer has a WriterExecState with key_points. None of them sees the others' state.

  • The Store for detailed data: complete results in ("research", task_id, agent_name) namespaces. The Supervisor reads summaries to decide; when it delegates, it pulls the complete data from the Store.

The Researcher at iteration 10 receives exactly the same volume of context as at iteration 1.


Troubleshooting

Problem 1: The supervisor's context window grows out of control

Symptom: The supervisor takes longer and longer to respond. The quality of its routing decisions degrades in later iterations.

Cause: Even though the workers have isolated state, the coordination state accumulates messages with each iteration. If you use add_messages, the messages from every round get concatenated.

Solution: Limit the supervisor's messages to the last N: state["messages"][-6:]. Use phase_results with summaries instead of storing complete outputs in messages.

Problem 2: The subgraph doesn't receive the state it expects

Symptom: A KeyError or None values when the subgraph tries to access its state fields.

Cause: The parent doesn't initialize every required field of the child's TypedDict.

Solution: Pass every field with default values when invoking the subgraph. Don't forget to initialize lists as [] and strings as "".

Problem 3: The Store has no data when the agent looks for it

Symptom: store.get() returns None even though the previous agent should have written data.

Cause: The namespace or the key doesn't match exactly. A typo in the namespace, or the config's task_id differs between invocations.

Solution: Centralize the namespace construction in an agent_namespace(task_id, agent_name) helper that returns ("tasks", task_id, agent_name). Use that function for both store.put() and store.get() — that eliminates typos.

Problem 4: Inbox messages don't get cleared between iterations

Symptom: In message passing, the analyst processes messages from previous iterations along with the current one.

Cause: The inbox accumulates messages with no timestamp or "already read" mechanism.

Solution: Add a read field to AgentMessage and filter by not m.read, or clear the inbox at the start of each iteration. Alternative: use timestamps and filter for messages after the last read.

Problem 5: The hybrid approach duplicates data

Symptom: The summary in phase_results says one thing, but the complete data in the Store says another.

Cause: The summary gets generated at one moment and the Store's data gets updated later (or vice versa).

Solution: Generate the summary from the same output you save in the Store. A single source of truth: full_output = result["messages"][-1].content, then store.put(...) with full_output and phase_results with full_output[:500].


Exercises

Exercise 1: Diagnose context bloat

You have this system with shared state:

class State(TypedDict):
    messages: Annotated[list, add_messages]
    research: str
    analysis: str
    draft: str
    feedback: str
    quality: float

def researcher(state):
    response = model.invoke([
        SystemMessage(content=f"""Find information.
Current draft: {state.get('draft', '')}
Feedback: {state.get('feedback', '')}
Quality: {state.get('quality', 0)}"""),
        *state["messages"],
    ])
    return {"messages": [response], "research": response.content}

Which fields shouldn't the researcher see? How would you fix it?

See solution

The researcher shouldn't see draft, feedback, or quality. Those fields belong to the writer and the supervisor. Seeing the draft makes the researcher look for information to "improve the report" instead of finding new information. The fix is to use isolated state:

class ResearcherState(TypedDict):
    messages: Annotated[list, add_messages]
    search_query: str

def researcher(state: ResearcherState):
    response = model.invoke([
        SystemMessage(content=f"Find information about: {state['search_query']}"),
        *state["messages"],
    ])
    return {"messages": [response]}

The parent extracts only the query from the coordination state and passes it to the researcher.

Exercise 2: Implement message passing with feedback

Implement a system where the supervisor can send a "feedback"-type message to the researcher asking for more data, without that feedback reaching the writer.

See solution
def supervisor_feedback(state: FeedbackState) -> dict:
    inbox = state.get("agent_inbox", {})
    inbox.setdefault("researcher", []).append(AgentMessage(
        sender="supervisor", receiver="researcher",
        content="I need more academic sources, not just blogs.",
        message_type="feedback",
    ))
    return {"agent_inbox": inbox}

def researcher_node(state: FeedbackState) -> dict:
    my_inbox = state.get("agent_inbox", {}).get("researcher", [])
    feedback = [m.content for m in my_inbox if m.message_type == "feedback"]

    prompt = "Find relevant information."
    if feedback:
        prompt += f"\nFeedback from the supervisor: {'; '.join(feedback)}"

    response = model.invoke([SystemMessage(content=prompt), *state["messages"]])

    inbox = state.get("agent_inbox", {})
    inbox.setdefault("analyst", []).append(AgentMessage(
        sender="researcher", receiver="analyst",
        content=response.content, message_type="result",
    ))
    return {"messages": [response], "agent_inbox": inbox}

The writer never sees the feedback because it only reads inbox["writer"]. The communication is directed — the supervisor sends feedback only to the researcher.

Exercise 3: Shared memory with the Store

Implement a system where the researcher writes results into the Store with a per-task namespace, and the analyst reads only the most recent results.

See solution
def researcher_store(state: dict, config: RunnableConfig, *, store: BaseStore):
    task_id = config["configurable"]["task_id"]
    timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
    response = model.invoke([
        SystemMessage(content="Find relevant information."), *state["messages"],
    ])
    store.put(("tasks", task_id, "researcher"), f"results_{timestamp}",
              {"data": response.content, "timestamp": timestamp})
    return {"messages": [response]}

def analyst_store(state: dict, config: RunnableConfig, *, store: BaseStore):
    task_id = config["configurable"]["task_id"]
    all_results = store.search(("tasks", task_id, "researcher"))
    if not all_results:
        return {"messages": [AIMessage(content="No data.")]}

    latest = sorted(all_results, key=lambda r: r.value.get("timestamp", ""), reverse=True)[0]
    response = model.invoke([
        SystemMessage(content=f"Analyze: {latest.value['data']}"),
        HumanMessage(content="Produce an analysis."),
    ])
    return {"messages": [response]}

The analyst uses store.search and takes the most recent result by timestamp.

Exercise 4: Design a hybrid state

Design the coordination state TypedDict and the execution states for a 4-agent system: researcher, analyst, writer, reviewer. The reviewer evaluates the draft and can ask the researcher to find more data or the writer to rewrite.

See solution
class CoordinationState(TypedDict):
    messages: Annotated[list, add_messages]
    current_phase: Literal["research", "analysis", "writing", "review", "done"]
    assigned_agent: str
    iteration_count: int
    quality_score: float
    phase_results: dict[str, str]
    reviewer_feedback: str  # so the supervisor knows what the reviewer asked for

class ResearcherExec(TypedDict):
    messages: Annotated[list, add_messages]
    search_query: str
    previous_feedback: str  # the reviewer's feedback if there's a re-search

class AnalystExec(TypedDict):
    messages: Annotated[list, add_messages]
    raw_data: str

class WriterExec(TypedDict):
    messages: Annotated[list, add_messages]
    key_points: list[str]
    rewrite_instructions: str  # the reviewer's instructions if there's a rewrite

class ReviewerExec(TypedDict):
    messages: Annotated[list, add_messages]
    draft_to_review: str
    evaluation_criteria: list[str]

The supervisor reads reviewer_feedback and decides whether to send it back to the researcher (more data) or the writer (a rewrite). The reviewer receives only the draft and the criteria — it shares state with nobody.

Exercise 5: Migrate from shared to hybrid

You have this pure shared-state system that works but suffers from context bloat at iteration 5+:

class FullSharedState(TypedDict):
    messages: Annotated[list, add_messages]
    research_data: str
    analysis: str
    draft: str
    quality: float
    iteration: int
    sources: list[str]
    feedback_history: list[str]

Migrate it to a hybrid approach while keeping the same functional flow.

See solution

Split it into coordination + execution:

class CoordState(TypedDict):
    messages: Annotated[list, add_messages]
    quality: float
    iteration: int
    phase_summaries: dict[str, str]
    current_agent: str

class ResearchExec(TypedDict):
    messages: Annotated[list, add_messages]
    query: str
    previous_sources: list[str]

class AnalysisExec(TypedDict):
    messages: Annotated[list, add_messages]
    data: str

class WritingExec(TypedDict):
    messages: Annotated[list, add_messages]
    points: list[str]
    feedback: str

def delegate(state: CoordState) -> dict:
    agent = state["current_agent"]
    sub = {"researcher": researcher_sub, "analyst": analyst_sub, "writer": writer_sub}[agent]

    if agent == "researcher":
        invoke_args = {"messages": [HumanMessage(content=state["messages"][-1].content)],
                       "query": state["messages"][-1].content, "previous_sources": []}
    elif agent == "analyst":
        invoke_args = {"messages": [HumanMessage(content="Analyze")],
                       "data": state.get("phase_summaries", {}).get("researcher", "")}
    else:
        invoke_args = {"messages": [HumanMessage(content="Write the report")],
                       "points": [state.get("phase_summaries", {}).get("analyst", "")],
                       "feedback": state["messages"][-1].content if state["iteration"] > 1 else ""}

    result = sub.invoke(invoke_args)
    output = result["messages"][-1].content
    return {"phase_summaries": {**state.get("phase_summaries", {}), agent: output[:500]}}

The research_data, analysis, draft, sources, and feedback_history fields disappear from the coordination state. The coordination state keeps phase_summaries with summarized versions for the supervisor's decisions without loading the full context.


Summary

In this capsule you learned:

  • Shared state — one shared TypedDict where everyone reads and writes. Simple but it causes context bloat: each agent sees information it doesn't need, degrading quality and driving up costs.

  • Isolated state — each agent has its own TypedDict and works in an independent subgraph. Constant context, no interference. The parent has to act as a translator between agents.

  • Message passing — selective communication with directed messages, not broadcast. Useful for feedback loops and non-linear communication.

  • The LangGraph Store — shared memory with per-agent namespaces. It persists between invocations and enables auditing.

  • Hybrid — the recommended approach: a shared coordination state + isolated execution states. It combines simple coordination with clean context.

  • The communication decision between agents is the most important one in the design. Pure shared state works in demos but collapses in production. Pure isolation is clean but hard to coordinate. Hybrid is the sweet spot.

Next capsule: Advanced orchestration — composing patterns (Router + Supervisor + Subagents), parallel execution, and how to combine everything you've learned in this module for the final project.


Additional Resources

  1. LangGraph State Management — Official documentation on how state works in LangGraph, reducers, and TypedDict
  2. LangGraph Multi-Agent — Communication — Communication patterns between agents: shared state vs message passing
  3. LangGraph Store API — The Store for shared memory between agents with namespaces
  4. LangGraph Subgraphs — How to implement subgraphs with state mapping for isolated execution
  5. Multi-Agent Architectures (LangChain Blog) — An analysis of orchestration patterns and state design decisions