Shared vs Isolated State

Capsule overview

This is the HARDEST design decision in multi-agent systems: what can each agent see and modify? Share too much and the agents interfere with each other — one agent overwrites another's work, state grows out of control, and debugging becomes impossible. Share too little and the agents can't coordinate — each one works inside its own bubble with no access to the context it needs.

Getting the state design right determines whether your system works or creates chaos. That's not an exaggeration: most bugs in multi-agent systems aren't logic errors inside the individual agents, they're problems with how those agents share (or fail to share) information.

In the previous capsules you used shared state without thinking about it much — every agent read from and wrote to the same TypedDict. Now you're going to understand why that works for small systems but breaks for large ones, and you'll learn three concrete patterns for designing state: shared, isolated, and hybrid.

The microservices analogy

If you come from backend development, this analogy will make everything click:

In microservices, sharing one database across services is a well-known anti-pattern: it produces coupling, conflicts, and makes it impossible to scale services independently. The same logic applies to agents.

The difference: in microservices, separating databases requires real infrastructure. In LangGraph, separating state is a design decision in your TypedDict. It's easier to implement, but just as important.

Pattern 1: Shared state — everyone sees everything

The simplest pattern: a single TypedDict, every agent reads and writes the same fields.

Implementation

from dotenv import load_dotenv
load_dotenv()

from typing import TypedDict, Annotated
import operator
from langgraph.graph import StateGraph, START, END
from langchain.chat_models import init_chat_model
from langchain_core.messages import AnyMessage, HumanMessage, SystemMessage
from IPython.display import Image, display

class SharedState(TypedDict):
    messages: Annotated[list[AnyMessage], operator.add]
    research_results: Annotated[list[str], operator.add]
    analysis: str
    final_report: str
    current_agent: str

model = init_chat_model("openai:gpt-4.1-mini")

def researcher(state: SharedState) -> dict:
    response = model.invoke([
        SystemMessage(content="You are a researcher. Find 3 key facts about the topic."),
        state["messages"][-1],
    ])
    findings = [f.strip() for f in response.content.split("\n") if f.strip()]
    return {
        "research_results": findings,
        "current_agent": "analyst",
    }

def analyst(state: SharedState) -> dict:
    findings_text = "\n".join(state["research_results"])
    response = model.invoke(
        f"Analyze these findings and extract the main conclusion:\n{findings_text}"
    )
    return {
        "analysis": response.content,
        "current_agent": "writer",
    }

def writer(state: SharedState) -> dict:
    response = model.invoke(
        f"Write an executive summary based on:\n"
        f"Findings: {', '.join(state['research_results'][:3])}\n"
        f"Analysis: {state['analysis']}"
    )
    return {
        "final_report": response.content,
        "current_agent": "done",
    }

graph = StateGraph(SharedState)
graph.add_node("researcher", researcher)
graph.add_node("analyst", analyst)
graph.add_node("writer", writer)
graph.add_edge(START, "researcher")
graph.add_edge("researcher", "analyst")
graph.add_edge("analyst", "writer")
graph.add_edge("writer", END)

app = graph.compile()
display(Image(app.get_graph().draw_mermaid_png()))

result = app.invoke({
    "messages": [HumanMessage(content="Impact of AI on education")],
    "research_results": [],
    "analysis": "",
    "final_report": "",
    "current_agent": "researcher",
})

print(f"Research results: {len(result['research_results'])} items")
print(f"Analysis: {result['analysis'][:100]}...")
print(f"Report: {result['final_report'][:100]}...")
# Output:
# Research results: 3 items
# Analysis: AI is transforming education across three main dimensions...
# Report: Executive summary: The integration of AI in education shows impact...

When it works

• ✅ Small systems (2-4 agents)
• ✅ The agents work in sequence (one finishes before the next starts)
• ✅ Each agent needs context from the previous ones (the analyst needs to see research_results)
• ✅ The execution order is fixed and predictable

The risk: interference between agents

The problem shows up when two agents modify the same field. Look at what happens if two researchers work in parallel:

from dotenv import load_dotenv
load_dotenv()

from typing import TypedDict
from langgraph.graph import StateGraph, START, END

class DangerousState(TypedDict):
    topic: str
    findings: str

def researcher_a(state: DangerousState) -> dict:
    return {"findings": "Finding A: AI improves personalized learning"}

def researcher_b(state: DangerousState) -> dict:
    return {"findings": "Finding B: AI reduces education costs"}

graph = StateGraph(DangerousState)
graph.add_node("researcher_a", researcher_a)
graph.add_node("researcher_b", researcher_b)

graph.add_edge(START, "researcher_a")
graph.add_edge(START, "researcher_b")
graph.add_edge("researcher_a", END)
graph.add_edge("researcher_b", END)

app = graph.compile()
result = app.invoke({"topic": "AI in education", "findings": ""})
print(result["findings"])
# Output: Finding B: AI reduces education costs
# Finding A is GONE! Last writer wins.

Without a reducer, findings gets overwritten. If both nodes run "in parallel" (LangGraph executes both in the same superstep), the result depends on the internal execution order. This is exactly the same problem as a race condition on a shared database.

Mitigation: reducers for safe accumulation

from typing import TypedDict, Annotated
import operator

class SafeSharedState(TypedDict):
    topic: str
    findings: Annotated[list[str], operator.add]

def researcher_a(state: SafeSharedState) -> dict:
    return {"findings": ["Finding A: AI improves personalized learning"]}

def researcher_b(state: SafeSharedState) -> dict:
    return {"findings": ["Finding B: AI reduces education costs"]}

With operator.add, both findings accumulate in the list. The reducer removes the overwrite conflict.

Pattern 2: Isolated state — each agent has its own scope

Each agent operates with its own TypedDict. An agent's state isn't directly visible to other agents. Communication goes through the parent node (supervisor or router), which maps data between scopes.

Implementation with a subgraph

from dotenv import load_dotenv
load_dotenv()

from typing import TypedDict, Annotated
import operator
from langgraph.graph import StateGraph, START, END
from langchain.chat_models import init_chat_model
from langchain_core.messages import AnyMessage, HumanMessage, SystemMessage
from IPython.display import Image, display

class ResearchSubState(TypedDict):
    query: str
    sources: Annotated[list[str], operator.add]
    findings: Annotated[list[str], operator.add]

class AnalysisSubState(TypedDict):
    input_data: list[str]
    conclusion: str
    confidence: float

class ParentState(TypedDict):
    messages: Annotated[list[AnyMessage], operator.add]
    research_output: list[str]
    analysis_conclusion: str
    analysis_confidence: float
    final_report: str

model = init_chat_model("openai:gpt-4.1-mini")

def research_node_1(state: ResearchSubState) -> dict:
    response = model.invoke(
        f"Find 2 sources about: {state['query']}. List the sources separated by '|'."
    )
    sources = [s.strip() for s in response.content.split("|")]
    return {"sources": sources}

def research_node_2(state: ResearchSubState) -> dict:
    response = model.invoke(
        f"Based on these sources: {', '.join(state['sources'])}\n"
        f"Extract the main findings about: {state['query']}"
    )
    findings = [f.strip() for f in response.content.split("\n") if f.strip()]
    return {"findings": findings}

research_graph = StateGraph(ResearchSubState)
research_graph.add_node("find_sources", research_node_1)
research_graph.add_node("extract_findings", research_node_2)
research_graph.add_edge(START, "find_sources")
research_graph.add_edge("find_sources", "extract_findings")
research_graph.add_edge("extract_findings", END)
research_subgraph = research_graph.compile()

def analysis_node(state: AnalysisSubState) -> dict:
    data_text = "\n".join(state["input_data"])
    response = model.invoke(
        f"Analyze this data and give a conclusion with a confidence level (0-1):\n{data_text}"
    )
    return {"conclusion": response.content, "confidence": 0.85}

analysis_graph = StateGraph(AnalysisSubState)
analysis_graph.add_node("analyze", analysis_node)
analysis_graph.add_edge(START, "analyze")
analysis_graph.add_edge("analyze", END)
analysis_subgraph = analysis_graph.compile()

def run_research(state: ParentState) -> dict:
    query = state["messages"][-1].content
    result = research_subgraph.invoke({
        "query": query,
        "sources": [],
        "findings": [],
    })
    return {"research_output": result["findings"]}

def run_analysis(state: ParentState) -> dict:
    result = analysis_subgraph.invoke({
        "input_data": state["research_output"],
        "conclusion": "",
        "confidence": 0.0,
    })
    return {
        "analysis_conclusion": result["conclusion"],
        "analysis_confidence": result["confidence"],
    }

def write_report(state: ParentState) -> dict:
    response = model.invoke(
        f"Generate an executive report:\n"
        f"Findings: {state['research_output'][:3]}\n"
        f"Conclusion: {state['analysis_conclusion']}\n"
        f"Confidence: {state['analysis_confidence']}"
    )
    return {"final_report": response.content}

parent_graph = StateGraph(ParentState)
parent_graph.add_node("research", run_research)
parent_graph.add_node("analysis", run_analysis)
parent_graph.add_node("report", write_report)
parent_graph.add_edge(START, "research")
parent_graph.add_edge("research", "analysis")
parent_graph.add_edge("analysis", "report")
parent_graph.add_edge("report", END)

app = parent_graph.compile()
display(Image(app.get_graph().draw_mermaid_png()))

result = app.invoke({
    "messages": [HumanMessage(content="Current state of quantum computing")],
    "research_output": [],
    "analysis_conclusion": "",
    "analysis_confidence": 0.0,
    "final_report": "",
})

print(f"Research findings: {len(result['research_output'])} items")
print(f"Analysis confidence: {result['analysis_confidence']}")
print(f"Report: {result['final_report'][:120]}...")
# Output:
# Research findings: 3 items
# Analysis confidence: 0.85
# Report: Executive Report: Quantum computing is currently in a phase of...

The key to the pattern

Notice how the states are completely separate:

• ResearchSubState has query, sources, findings — fields only the researcher needs
• AnalysisSubState has input_data, conclusion, confidence — fields only the analyst needs
• ParentState has each agent's final results — it's the "contract" between agents

The researcher can't see conclusion. The analyst can't see sources. Each agent operates in its own world, and the parent maps the data between them:

research.findings → parent.research_output → analysis.input_data

When to use isolated state

• ✅ The agents can work in parallel with no dependencies
• ✅ Each agent has internal data the others don't need to see
• ✅ The system is large (5+ agents) and shared state would be a mess
• ✅ You need each agent to be testable independently

The risk: mapping overhead

Every time you want to pass data between agents, you need explicit mapping code. If you have 5 agents sharing results in different combinations, the mapping in the parent gets tedious.

Pattern 3: Hybrid — shared to coordinate, isolated to work

This is the recommended pattern for most production systems. Define a minimal shared state for coordination and let each agent keep private fields for its internal work.

State design

from typing import TypedDict, Annotated
import operator
from langchain_core.messages import AnyMessage

class HybridState(TypedDict):
    # --- SHARED: every agent sees and uses these ---
    messages: Annotated[list[AnyMessage], operator.add]
    task_status: str
    current_phase: str

    # --- OUTPUTS from each agent (shared as results) ---
    research_results: Annotated[list[str], operator.add]
    analysis_summary: str
    final_report: str

    # --- ISOLATED by convention: only the agent that "owns" it uses it ---
    _research_sources: Annotated[list[str], operator.add]
    _research_queries: Annotated[list[str], operator.add]
    _analysis_metrics: dict
    _writer_drafts: Annotated[list[str], operator.add]

The _prefix convention marks fields that "belong" to a specific agent. It isn't technically enforced (any agent CAN read _research_sources), but it's a team contract: "this field is the researcher's responsibility, don't touch it."

Full implementation

from dotenv import load_dotenv
load_dotenv()

from typing import TypedDict, Annotated
import operator
from langgraph.graph import StateGraph, START, END
from langchain.chat_models import init_chat_model
from langchain_core.messages import AnyMessage, HumanMessage, SystemMessage
from IPython.display import Image, display

class TeamState(TypedDict):
    messages: Annotated[list[AnyMessage], operator.add]
    task_status: str
    current_phase: str
    research_results: Annotated[list[str], operator.add]
    analysis_summary: str
    final_report: str
    _research_sources: Annotated[list[str], operator.add]
    _analysis_raw_scores: Annotated[list[float], operator.add]

model = init_chat_model("openai:gpt-4.1-mini")

def researcher(state: TeamState) -> dict:
    topic = state["messages"][-1].content
    response = model.invoke([
        SystemMessage(content="Find 3 key facts. Separate them with '|'."),
        HumanMessage(content=topic),
    ])
    findings = [f.strip() for f in response.content.split("|") if f.strip()]
    sources = ["web_search", "academic_db", "news_api"]

    return {
        "research_results": findings,
        "_research_sources": sources,
        "task_status": "research_complete",
        "current_phase": "analysis",
    }

def analyst(state: TeamState) -> dict:
    findings_text = "\n".join(state["research_results"])
    response = model.invoke(
        f"Analyze these findings. Give a 2-sentence summary:\n{findings_text}"
    )
    return {
        "analysis_summary": response.content,
        "_analysis_raw_scores": [0.8, 0.75, 0.9],
        "task_status": "analysis_complete",
        "current_phase": "writing",
    }

def writer(state: TeamState) -> dict:
    response = model.invoke(
        f"Write a 3-paragraph executive report based on:\n"
        f"Findings: {state['research_results']}\n"
        f"Analysis: {state['analysis_summary']}"
    )
    return {
        "final_report": response.content,
        "task_status": "complete",
        "current_phase": "done",
    }

graph = StateGraph(TeamState)
graph.add_node("researcher", researcher)
graph.add_node("analyst", analyst)
graph.add_node("writer", writer)
graph.add_edge(START, "researcher")
graph.add_edge("researcher", "analyst")
graph.add_edge("analyst", "writer")
graph.add_edge("writer", END)

app = graph.compile()
display(Image(app.get_graph().draw_mermaid_png()))

result = app.invoke({
    "messages": [HumanMessage(content="Trends in renewable energy 2025")],
    "task_status": "started",
    "current_phase": "research",
    "research_results": [],
    "analysis_summary": "",
    "final_report": "",
    "_research_sources": [],
    "_analysis_raw_scores": [],
})

print(f"Status: {result['task_status']}")
print(f"Phase: {result['current_phase']}")
print(f"Sources (private to researcher): {result['_research_sources']}")
print(f"Raw scores (private to analyst): {result['_analysis_raw_scores']}")
print(f"Research results (shared): {len(result['research_results'])} items")
print(f"Report (first 100 chars): {result['final_report'][:100]}...")
# Output:
# Status: complete
# Phase: done
# Sources (private to researcher): ['web_search', 'academic_db', 'news_api']
# Raw scores (private to analyst): [0.8, 0.75, 0.9]
# Research results (shared): 3 items
# Report (first 100 chars): Executive Report: Renewable energy continues its global expansion...

Why hybrid is the recommended default

1. Visible coordination: task_status and current_phase are available to everyone — any agent can know which step we're on
2. Accessible results: research_results and analysis_summary are shared outputs — the writer needs to see both
3. Protected internals: _research_sources and _analysis_raw_scores are working data that only matter to the agent that owns them
4. Clear debugging: you can inspect the final state and know exactly what each agent produced

Designing the state contract

Before you write a single node, ask yourself these three questions for each agent:

1. What does this agent NEED to see? (minimum inputs)

# The researcher needs:
#   - messages (to know what to research)
#   - current_phase (to know if it's its turn)
# It does NOT need: analysis_summary, final_report, _analysis_raw_scores

# The analyst needs:
#   - research_results (to analyze)
# It does NOT need: _research_sources, _research_queries

# The writer needs:
#   - research_results + analysis_summary (to write)
# It does NOT need: _research_sources, _analysis_raw_scores

2. What does this agent PRODUCE? (outputs)

# The researcher produces:
#   - research_results (shared, others need it)
#   - _research_sources (private, debugging only)

# The analyst produces:
#   - analysis_summary (shared, the writer needs it)
#   - _analysis_raw_scores (private, debugging only)

# The writer produces:
#   - final_report (shared, it's the system's output)

3. How are conflicts resolved if two agents write to the same field?

# research_results: operator.add → accumulates (multiple researchers can contribute)
# analysis_summary: no reducer → last analyst wins (there's only one)
# task_status: no reducer → the last one to run defines the current status

This analysis is equivalent to designing API interfaces between microservices. Each agent has a "contract": what it receives, what it returns, and what guarantees it offers. Spend time on this BEFORE writing code.

Preventing conflicts: parallel execution and state

When two agents run in parallel (the same superstep in LangGraph), state conflicts are real. Let's look at the strategies:

Strategy 1: Reducers for safe accumulation

If two agents write to the same list-typed field, operator.add merges them:

from dotenv import load_dotenv
load_dotenv()

from typing import TypedDict, Annotated
import operator
from langgraph.graph import StateGraph, START, END

class ParallelState(TypedDict):
    topic: str
    findings: Annotated[list[str], operator.add]

def web_researcher(state: ParallelState) -> dict:
    return {"findings": [f"[Web] Fact about {state['topic']}"]}

def academic_researcher(state: ParallelState) -> dict:
    return {"findings": [f"[Academic] Study on {state['topic']}"]}

def news_researcher(state: ParallelState) -> dict:
    return {"findings": [f"[News] News item about {state['topic']}"]}

graph = StateGraph(ParallelState)
graph.add_node("web", web_researcher)
graph.add_node("academic", academic_researcher)
graph.add_node("news", news_researcher)

graph.add_edge(START, "web")
graph.add_edge(START, "academic")
graph.add_edge(START, "news")
graph.add_edge("web", END)
graph.add_edge("academic", END)
graph.add_edge("news", END)

app = graph.compile()
result = app.invoke({"topic": "generative AI", "findings": []})
print(result["findings"])
# Output: ['[Web] Fact about generative AI', '[Academic] Study on generative AI', '[News] News item about generative AI']

Three agents running in parallel, all writing to findings. Thanks to operator.add, the three results accumulate with no conflicts.

Strategy 2: Separate fields per agent

If each agent produces data of a different nature, use separate fields:

from typing import TypedDict, Annotated
import operator
from langchain_core.messages import AnyMessage

class SeparatedState(TypedDict):
    messages: Annotated[list[AnyMessage], operator.add]
    web_results: Annotated[list[str], operator.add]
    academic_results: Annotated[list[str], operator.add]
    news_results: Annotated[list[str], operator.add]

Zero conflicts, because each agent writes to its own field. The next node can read all three fields to combine them.

Strategy 3: Custom reducer with a timestamp

For scalar fields that several agents might update, a reducer with a timestamp preserves the most recent one:

import time

def latest_wins(current: dict, new: dict) -> dict:
    """Keeps the value with the most recent timestamp."""
    if not current or new.get("timestamp", 0) > current.get("timestamp", 0):
        return new
    return current

class TimestampedState(TypedDict):
    status: Annotated[dict, latest_wins]

Communication between agents: message passing through state

Agents don't call each other directly. They communicate by writing and reading state fields. This is implicit message passing.

Pattern: results that flow from one agent to the next

from dotenv import load_dotenv
load_dotenv()

from typing import TypedDict, Annotated
import operator
from langgraph.graph import StateGraph, START, END
from langchain.chat_models import init_chat_model
from langchain_core.messages import AnyMessage, HumanMessage, SystemMessage
from IPython.display import Image, display

class PipelineState(TypedDict):
    messages: Annotated[list[AnyMessage], operator.add]
    raw_data: str
    cleaned_data: str
    analysis: str
    recommendations: Annotated[list[str], operator.add]

model = init_chat_model("openai:gpt-4.1-mini")

def data_collector(state: PipelineState) -> dict:
    """Collects raw data. Writes to: raw_data."""
    topic = state["messages"][-1].content
    response = model.invoke(f"Generate sample data about: {topic}. Format: a list of 3 metrics.")
    return {"raw_data": response.content}

def data_cleaner(state: PipelineState) -> dict:
    """Reads from: raw_data. Writes to: cleaned_data."""
    response = model.invoke(
        f"Clean and structure this data:\n{state['raw_data']}"
    )
    return {"cleaned_data": response.content}

def data_analyst(state: PipelineState) -> dict:
    """Reads from: cleaned_data. Writes to: analysis, recommendations."""
    response = model.invoke(
        f"Analyze this data and give 2 recommendations:\n{state['cleaned_data']}"
    )
    return {
        "analysis": response.content,
        "recommendations": ["Recommendation based on the analysis"],
    }

graph = StateGraph(PipelineState)
graph.add_node("collector", data_collector)
graph.add_node("cleaner", data_cleaner)
graph.add_node("analyst", data_analyst)
graph.add_edge(START, "collector")
graph.add_edge("collector", "cleaner")
graph.add_edge("cleaner", "analyst")
graph.add_edge("analyst", END)

app = graph.compile()
display(Image(app.get_graph().draw_mermaid_png()))

result = app.invoke({
    "messages": [HumanMessage(content="Web traffic for the last month")],
    "raw_data": "",
    "cleaned_data": "",
    "analysis": "",
    "recommendations": [],
})

print(f"Raw data: {result['raw_data'][:80]}...")
print(f"Cleaned: {result['cleaned_data'][:80]}...")
print(f"Analysis: {result['analysis'][:80]}...")
print(f"Recommendations: {result['recommendations']}")
# Output:
# Raw data: 1. Total visits: 45,000  2. Bounce rate: 62%  3. Average time...
# Cleaned: Web traffic metrics: Visits: 45,000 | Bounce rate: 62% | Time...
# Analysis: The analysis reveals a high bounce rate (62%) that suggests problems...
# Recommendations: ['Recommendation based on the analysis']

The data flow is explicit:

collector → raw_data → cleaner → cleaned_data → analyst → analysis + recommendations

Each agent reads the fields it needs and writes the fields it produces. There are no direct calls between agents — everything goes through the state.

Full example: a hybrid system with 4 agents

This example brings it all together: shared state for coordination, dedicated fields per agent, reducers for accumulation, and a parallel + sequential flow.

from dotenv import load_dotenv
load_dotenv()

from typing import TypedDict, Annotated
import operator
from langgraph.graph import StateGraph, START, END
from langchain.chat_models import init_chat_model
from langchain_core.messages import AnyMessage, HumanMessage, SystemMessage
from IPython.display import Image, display

class ProductionState(TypedDict):
    messages: Annotated[list[AnyMessage], operator.add]
    phase: str
    task_description: str

    web_findings: Annotated[list[str], operator.add]
    paper_findings: Annotated[list[str], operator.add]

    all_findings: Annotated[list[str], operator.add]
    analysis: str
    report: str

model = init_chat_model("openai:gpt-4.1-mini")

def web_researcher(state: ProductionState) -> dict:
    response = model.invoke([
        SystemMessage(content="Search the web. Give 2 findings separated by '|'."),
        HumanMessage(content=state["task_description"]),
    ])
    findings = [f"[Web] {f.strip()}" for f in response.content.split("|") if f.strip()]
    return {
        "web_findings": findings,
        "all_findings": findings,
    }

def paper_researcher(state: ProductionState) -> dict:
    response = model.invoke([
        SystemMessage(content="Search academic papers. Give 2 findings separated by '|'."),
        HumanMessage(content=state["task_description"]),
    ])
    findings = [f"[Paper] {f.strip()}" for f in response.content.split("|") if f.strip()]
    return {
        "paper_findings": findings,
        "all_findings": findings,
    }

def analyst(state: ProductionState) -> dict:
    all_data = "\n".join(state["all_findings"])
    response = model.invoke(
        f"Analyze all the findings and synthesize them into a 2-3 sentence conclusion:\n{all_data}"
    )
    return {
        "analysis": response.content,
        "phase": "writing",
    }

def writer(state: ProductionState) -> dict:
    response = model.invoke(
        f"Write a 3-paragraph executive report based on:\n"
        f"Web findings: {state['web_findings']}\n"
        f"Academic findings: {state['paper_findings']}\n"
        f"Analysis: {state['analysis']}"
    )
    return {
        "report": response.content,
        "phase": "complete",
    }

graph = StateGraph(ProductionState)
graph.add_node("web_researcher", web_researcher)
graph.add_node("paper_researcher", paper_researcher)
graph.add_node("analyst", analyst)
graph.add_node("writer", writer)

graph.add_edge(START, "web_researcher")
graph.add_edge(START, "paper_researcher")
graph.add_edge("web_researcher", "analyst")
graph.add_edge("paper_researcher", "analyst")
graph.add_edge("analyst", "writer")
graph.add_edge("writer", END)

app = graph.compile()
display(Image(app.get_graph().draw_mermaid_png()))

result = app.invoke({
    "messages": [HumanMessage(content="Research AI in medicine")],
    "phase": "research",
    "task_description": "Impact of artificial intelligence on medical diagnosis",
    "web_findings": [],
    "paper_findings": [],
    "all_findings": [],
    "analysis": "",
    "report": "",
})

print(f"Phase: {result['phase']}")
print(f"Web findings: {len(result['web_findings'])} items")
print(f"Paper findings: {len(result['paper_findings'])} items")
print(f"All findings: {len(result['all_findings'])} items")
print(f"Analysis: {result['analysis'][:100]}...")
print(f"Report: {result['report'][:100]}...")
# Output:
# Phase: complete
# Web findings: 2 items
# Paper findings: 2 items
# All findings: 4 items
# Analysis: AI is showing significant improvements in medical diagnosis...
# Report: Executive Report: The integration of artificial intelligence into...

The state design mirrors the architecture:

• web_findings and paper_findings: isolated fields, each researcher writes to its own
• all_findings: shared field with operator.add where both researchers accumulate
• analysis and report: sequential outputs, each one reads from the previous
• phase: global coordination, tells you which stage the system is in

Decision table: which pattern do you pick?

Troubleshooting

Problem 1: An agent reads a field another agent hasn't written yet

Symptom: KeyError: 'analysis_summary' in the writer, because the analyst hasn't run yet. Cause: The graph edges don't guarantee the right order, or the field has no default value. Fix: Check that the edges force the correct sequence, and always include initial values:

result = app.invoke({
    "analysis_summary": "",  # Default value
    "research_results": [],
    # ... every field with defaults
})

Problem 2: Data from parallel execution gets lost

Symptom: Two agents run in parallel writing to the same list-typed field, but only one agent's results show up. Cause: The field doesn't have Annotated[list, operator.add]. Fix: Add the reducer for fields that multiple agents write to:

class State(TypedDict):
    findings: list[str]  # BAD → the last agent overwrites

class State(TypedDict):
    findings: Annotated[list[str], operator.add]  # GOOD → they accumulate

Problem 3: State grows out of control in long loops

Symptom: After 20 iterations of a supervisor → agents loop, the state has thousands of entries in the accumulated lists, and every LLM call is slower and more expensive. Cause: operator.add accumulates indefinitely. In long loops, the lists grow without bound. Fix: Use a custom reducer that caps the size:

def capped_list(current: list, new: list, max_size: int = 50) -> list:
    combined = current + new
    return combined[-max_size:]

class State(TypedDict):
    findings: Annotated[list[str], lambda c, n: (c + n)[-50:]]

Problem 4: I don't know which agent wrote what into the state

Symptom: The final state has research_results with 8 items, but you don't know which ones came from which agent. Cause: Several agents accumulate into the same field without identifying themselves. Fix: Include origin metadata in each entry:

def researcher_a(state) -> dict:
    return {"findings": [{"source": "researcher_a", "data": "finding X"}]}

def researcher_b(state) -> dict:
    return {"findings": [{"source": "researcher_b", "data": "finding Y"}]}

Problem 5: The subgraph can't access the parent's state

Symptom: A compiled subgraph needs data from the parent state, but its TypedDict is different and has no access. Cause: Subgraphs have their own isolated state. That's the whole point. Fix: The parent node that invokes the subgraph has to explicitly map the data it needs:

def run_sub_agent(state: ParentState) -> dict:
    sub_result = sub_graph.invoke({
        "query": state["messages"][-1].content,
        "context": state["research_results"],
    })
    return {"analysis": sub_result["conclusion"]}

Exercises

Exercise 1: Identify the right pattern (Easy)

For each scenario, decide whether you'd use shared, isolated, or hybrid state. Justify it.

A) 3 agents in sequence: researcher → analyst → writer. Each one needs the previous one's output.

B) 5 translation agents that translate the same text into 5 languages in parallel.

C) A supervisor with 3 specialized agents that work in phases, but each agent has internal calculations the others don't need to see.

Exercise 2: Fix the state conflict (Easy)

The following code has a bug: two researchers run in parallel but only one of them survives in the results. Fix it.

from typing import TypedDict
from langgraph.graph import StateGraph, START, END

class BuggyState(TypedDict):
    topic: str
    results: list[str]

def researcher_a(state: BuggyState) -> dict:
    return {"results": [f"Result A about {state['topic']}"]}

def researcher_b(state: BuggyState) -> dict:
    return {"results": [f"Result B about {state['topic']}"]}

graph = StateGraph(BuggyState)
graph.add_node("a", researcher_a)
graph.add_node("b", researcher_b)
graph.add_edge(START, "a")
graph.add_edge(START, "b")
graph.add_edge("a", END)
graph.add_edge("b", END)

app = graph.compile()
result = app.invoke({"topic": "AI", "results": []})
print(result["results"])
# Actual: ['Result B about AI'] — Result A is gone!

from typing import TypedDict, Annotated
import operator
from langgraph.graph import StateGraph, START, END

class FixedState(TypedDict):
    topic: str
    results: Annotated[list[str], operator.add]  # ← Add the reducer

def researcher_a(state: FixedState) -> dict:
    return {"results": [f"Result A about {state['topic']}"]}

def researcher_b(state: FixedState) -> dict:
    return {"results": [f"Result B about {state['topic']}"]}

graph = StateGraph(FixedState)
graph.add_node("a", researcher_a)
graph.add_node("b", researcher_b)
graph.add_edge(START, "a")
graph.add_edge(START, "b")
graph.add_edge("a", END)
graph.add_edge("b", END)

app = graph.compile()
result = app.invoke({"topic": "AI", "results": []})
print(result["results"])
# Output: ['Result A about AI', 'Result B about AI']

Exercise 3: Design hybrid state for a team of 4 agents (Medium)

Design the TypedDict for a system of 4 agents: data_collector (collects data from APIs), validator (validates format and quality), transformer (transforms data), reporter (generates a report). Define which fields are shared, which are private, and which need reducers. You don't need to implement the agents.

from typing import TypedDict, Annotated
import operator
from langchain_core.messages import AnyMessage

class DataPipelineState(TypedDict):
    # SHARED: coordination
    messages: Annotated[list[AnyMessage], operator.add]
    current_phase: str
    error_count: int

    # SHARED OUTPUTS (each agent writes, the next one reads)
    raw_data: Annotated[list[dict], operator.add]
    validation_result: dict
    transformed_data: list[dict]
    final_report: str

    # PRIVATE to the collector
    _collector_api_calls: Annotated[list[str], operator.add]
    _collector_retries: int

    # PRIVATE to the validator
    _validation_errors: Annotated[list[str], operator.add]
    _validation_schema: str

    # PRIVATE to the transformer
    _transform_steps: Annotated[list[str], operator.add]
    _transform_duration_ms: float

    # PRIVATE to the reporter
    _report_drafts: Annotated[list[str], operator.add]

Exercise 4: Implement communication between agents (Medium)

Build a system of 3 agents where the planner writes a plan, the executor reads the plan and generates results, and the reviewer reads the results and decides whether the plan was fulfilled. Use the hybrid pattern. Implement it with StateGraph.

from dotenv import load_dotenv
load_dotenv()

from typing import TypedDict, Annotated
import operator
from langgraph.graph import StateGraph, START, END
from langchain.chat_models import init_chat_model
from langchain_core.messages import AnyMessage, HumanMessage, SystemMessage

class PlanExecuteState(TypedDict):
    messages: Annotated[list[AnyMessage], operator.add]
    phase: str

    plan: str
    execution_results: Annotated[list[str], operator.add]
    review: str
    plan_fulfilled: bool

    _planner_alternatives: Annotated[list[str], operator.add]
    _executor_attempts: int

model = init_chat_model("openai:gpt-4.1-mini")

def planner(state: PlanExecuteState) -> dict:
    task = state["messages"][-1].content
    response = model.invoke([
        SystemMessage(content="Create a 3-step plan for this task. Be concise."),
        HumanMessage(content=task),
    ])
    return {
        "plan": response.content,
        "phase": "executing",
        "_planner_alternatives": ["Alternative plan: a different approach"],
    }

def executor(state: PlanExecuteState) -> dict:
    response = model.invoke(
        f"Execute this plan and report the results of each step:\n{state['plan']}"
    )
    results = [r.strip() for r in response.content.split("\n") if r.strip()]
    return {
        "execution_results": results,
        "phase": "reviewing",
        "_executor_attempts": 1,
    }

def reviewer(state: PlanExecuteState) -> dict:
    response = model.invoke(
        f"Review whether the plan was fulfilled:\n"
        f"Plan: {state['plan']}\n"
        f"Results: {state['execution_results']}\n"
        f"Answer 'FULFILLED' or 'NOT FULFILLED' followed by a short explanation."
    )
    # Careful: a `"fulfilled" in response` check would also return True for "NOT FULFILLED",
    # because "fulfilled" is a substring of "not fulfilled". We anchor to the start.
    fulfilled = response.content.strip().upper().startswith("FULFILLED")
    return {
        "review": response.content,
        "plan_fulfilled": fulfilled,
        "phase": "complete",
    }

graph = StateGraph(PlanExecuteState)
graph.add_node("planner", planner)
graph.add_node("executor", executor)
graph.add_node("reviewer", reviewer)
graph.add_edge(START, "planner")
graph.add_edge("planner", "executor")
graph.add_edge("executor", "reviewer")
graph.add_edge("reviewer", END)

app = graph.compile()

result = app.invoke({
    "messages": [HumanMessage(content="Research the advantages of Python for data science")],
    "phase": "planning",
    "plan": "",
    "execution_results": [],
    "review": "",
    "plan_fulfilled": False,
    "_planner_alternatives": [],
    "_executor_attempts": 0,
})

print(f"Phase: {result['phase']}")
print(f"Plan: {result['plan'][:100]}...")
print(f"Execution results: {len(result['execution_results'])} items")
print(f"Review: {result['review'][:100]}...")
print(f"Plan fulfilled: {result['plan_fulfilled']}")
# Output:
# Phase: complete
# Plan: 1. Research the main Python libraries for data science...
# Execution results: 3 items
# Review: FULFILLED. The results cover the three steps of the plan...
# Plan fulfilled: True

Exercise 5: Subgraph with isolated state (Advanced)

Build a system where the parent graph has ParentState and a subgraph has ResearchSubState. The parent invokes the subgraph passing it only the data it needs, and gets back only the results. Show that the subgraph CANNOT see the parent's fields.

from dotenv import load_dotenv
load_dotenv()

from typing import TypedDict, Annotated
import operator
from langgraph.graph import StateGraph, START, END
from langchain.chat_models import init_chat_model
from langchain_core.messages import AnyMessage, HumanMessage

class ResearchSubState(TypedDict):
    query: str
    max_results: int
    findings: Annotated[list[str], operator.add]

class ParentState(TypedDict):
    messages: Annotated[list[AnyMessage], operator.add]
    budget: float
    deadline: str
    research_output: Annotated[list[str], operator.add]
    summary: str

model = init_chat_model("openai:gpt-4.1-mini")

def sub_search(state: ResearchSubState) -> dict:
    response = model.invoke(
        f"Find {state['max_results']} facts about: {state['query']}. Separate them with '|'."
    )
    findings = [f.strip() for f in response.content.split("|") if f.strip()]
    return {"findings": findings[:state["max_results"]]}

sub_graph_builder = StateGraph(ResearchSubState)
sub_graph_builder.add_node("search", sub_search)
sub_graph_builder.add_edge(START, "search")
sub_graph_builder.add_edge("search", END)
research_subgraph = sub_graph_builder.compile()

def invoke_research(state: ParentState) -> dict:
    query = state["messages"][-1].content
    sub_result = research_subgraph.invoke({
        "query": query,
        "max_results": 3,
        "findings": [],
    })
    return {"research_output": sub_result["findings"]}

def summarize(state: ParentState) -> dict:
    findings = "\n".join(state["research_output"])
    response = model.invoke(f"Summarize these findings in 2 sentences:\n{findings}")
    return {"summary": response.content}

parent_builder = StateGraph(ParentState)
parent_builder.add_node("research", invoke_research)
parent_builder.add_node("summarize", summarize)
parent_builder.add_edge(START, "research")
parent_builder.add_edge("research", "summarize")
parent_builder.add_edge("summarize", END)

app = parent_builder.compile()
result = app.invoke({
    "messages": [HumanMessage(content="Blockchain trends 2025")],
    "budget": 1000.0,
    "deadline": "2025-12-31",
    "research_output": [],
    "summary": "",
})

print(f"Budget (parent only): {result['budget']}")
print(f"Deadline (parent only): {result['deadline']}")
print(f"Research output: {len(result['research_output'])} items")
print(f"Summary: {result['summary'][:100]}...")
# Output:
# Budget (parent only): 1000.0
# Deadline (parent only): 2025-12-31
# Research output: 3 items
# Summary: Blockchain keeps evolving with trends toward...

Exercise 6: Custom reducer to merge parallel agents (Advanced)

Build a custom reducer that combines results from several parallel agents, prioritizing by confidence. Each agent returns a dict shaped like {"data": str, "confidence": float}. The reducer must keep a list sorted by descending confidence, capped at 5 items.

from dotenv import load_dotenv
load_dotenv()

from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, START, END

def priority_merge(current: list[dict], new: list[dict]) -> list[dict]:
    """Merges results by confidence, keeps the top 5."""
    combined = current + new
    sorted_results = sorted(combined, key=lambda x: x["confidence"], reverse=True)
    return sorted_results[:5]

class PriorityState(TypedDict):
    topic: str
    results: Annotated[list[dict], priority_merge]

def agent_high_confidence(state: PriorityState) -> dict:
    return {"results": [
        {"data": f"Precise finding about {state['topic']}", "confidence": 0.95},
        {"data": f"Verified fact on {state['topic']}", "confidence": 0.88},
    ]}

def agent_medium_confidence(state: PriorityState) -> dict:
    return {"results": [
        {"data": f"Partial information about {state['topic']}", "confidence": 0.65},
        {"data": f"General reference on {state['topic']}", "confidence": 0.55},
    ]}

def agent_low_confidence(state: PriorityState) -> dict:
    return {"results": [
        {"data": f"Unverified fact about {state['topic']}", "confidence": 0.30},
        {"data": f"Rumor about {state['topic']}", "confidence": 0.15},
    ]}

graph = StateGraph(PriorityState)
graph.add_node("high", agent_high_confidence)
graph.add_node("medium", agent_medium_confidence)
graph.add_node("low", agent_low_confidence)

graph.add_edge(START, "high")
graph.add_edge(START, "medium")
graph.add_edge(START, "low")
graph.add_edge("high", END)
graph.add_edge("medium", END)
graph.add_edge("low", END)

app = graph.compile()
result = app.invoke({"topic": "quantum computing", "results": []})

print(f"Total results (capped at 5): {len(result['results'])}")
for r in result["results"]:
    print(f"  [{r['confidence']:.2f}] {r['data']}")
# Output:
# Total results (capped at 5): 5
#   [0.95] Precise finding about quantum computing
#   [0.88] Verified fact on quantum computing
#   [0.65] Partial information about quantum computing
#   [0.55] General reference on quantum computing
#   [0.30] Unverified fact about quantum computing
# (The "Rumor" at 0.15 was dropped — only the top 5 survive)

Summary

In this capsule you learned:

• State design is the hardest decision in multi-agent — it determines whether your system works or creates chaos
• The microservices analogy clarifies the trade-offs: shared DB = shared state (dangerous), each service with its own DB = isolated state (safe), APIs between services = message passing
• Shared state (everyone sees everything) works for small, sequential systems, but produces conflicts under parallel execution
• Isolated state (each agent has its own scope) eliminates conflicts but requires explicit mapping between agents
• Hybrid state (shared to coordinate, isolated to work) is the recommended pattern for production
• Reducers (operator.add, custom) are the main tool for preventing overwrite conflicts
• Communication between agents is implicit message passing through the state: one agent writes, another reads
• Designing the state is designing contracts between agents: what each one receives, what it produces, how conflicts get resolved

Next capsule: the module project — integrating supervisor, router, handoffs, and hybrid state into a complete multi-agent system.

Additional resources

1. State Management — LangGraph Docs — Official reference for state and reducers
2. Multi-Agent Architectures — LangGraph Docs — Communication patterns between agents
3. How to pass private state between nodes — Official guide to private state
4. Subgraphs — LangGraph Docs — Documentation on subgraphs with isolated state
5. Reducers — LangGraph Docs — Detailed reference on reducers and Annotated
6. How to add and use subgraphs — Practical subgraph tutorial
7. State Schema — LangGraph How-tos — How to define state schemas for complex graphs

Module 10 — LangChain & LangGraph: From Chains to Agents