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:
-
Quality degradation. The LLM loses focus with irrelevant information. The researcher starts "responding" to the analyst's analysis instead of finding new information.
-
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.
-
Interference between agents. One agent reads fields another left in an inconsistent state. The classic bug: the researcher sees
report_draftand 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
- Selective access. The researcher can't read what the writer produces. There's no broadcast — each agent explicitly looks up what it needs.
- Persistence. The Store survives between invocations. If the system fails mid-run, the partial results are saved.
- Rich metadata. Timestamps, versions, quality scores alongside the data. The analyst can decide whether it needs a re-search.
- 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
| Aspect | Shared State | Isolated State | Message Passing | Hybrid |
|---|---|---|---|---|
| Implementation complexity | Low | Medium | Medium-High | High |
| Context bloat | Yes, grows with each iteration | No | Controlled by design | No |
| Coordination between agents | Trivial | Requires the parent as a translator | Explicit, via messages | Easy via the coordination state |
| Debugging | One state, easy | Each agent isolated, easy | Tracing messages, medium | State + Store, medium |
| Scalability (# agents) | Poor (>4 agents) | Good | Good | Good |
| Interference between agents | High | None | Low | Low |
| Data persistence | Only with a checkpointer | Lost when the subgraph ends | Lost when it ends | The Store persists |
| When to use | Prototypes, 2-3 agents | Decomposable tasks | Complex feedback loops | Production |
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
ResearcherExecStatewithsearch_queryandfound_sources. The Analyst has anAnalystExecStatewithraw_data. The Writer has aWriterExecStatewithkey_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
TypedDictwhere 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
TypedDictand 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
- LangGraph State Management — Official documentation on how state works in LangGraph, reducers, and TypedDict
- LangGraph Multi-Agent — Communication — Communication patterns between agents: shared state vs message passing
- LangGraph Store API — The Store for shared memory between agents with namespaces
- LangGraph Subgraphs — How to implement subgraphs with state mapping for isolated execution
- Multi-Agent Architectures (LangChain Blog) — An analysis of orchestration patterns and state design decisions