StateGraph Fundamentals: State, Reducers, Nodes, and Edges (LangGraph in Production, Module 02)

This is Module 02 of LangGraph in Production, a free 16-module course that takes you from your first StateGraph to a deployed, durable autonomous coding agent on LangGraph v1.x. Start at Module 1 or browse the full syllabus.

The problem: an agent with no memory

Picture this. You hand Forge an issue against Tally: “tally report raises ZeroDivisionError when there are no expenses.”

The Module 1 version of Forge can send that string to a model and get a reply. Then what? Where does it keep the original issue text so the next node can read it? Where does the agent’s chain of reasoning live between nodes? Where will the path to the sandbox copy of the repo go?

You open a print(state) inside a node and you see a plain Python dict. No contract. No guarantee that messages accumulates rather than overwrites. No type hints. The first time two nodes both write to messages, the second write silently clobbers the first — and Forge has already forgotten the question it was trying to answer.

An agent without a state schema is an agent that loses what it just learned. That is the douleur this module fixes. By the end of it, Forge owns ForgeState: a typed working memory whose every key knows how to handle concurrent writes — and whose structure will never change, only grow.

In this module

You’ll learn:

How to define a typed state schema (TypedDict, Pydantic, or @dataclass) and pass it to StateGraph.
What a channel is, and why an unannotated channel uses last-write-wins semantics.
How to attach a reducer to a channel with Annotated, and why add_messages is the right reducer for the agent scratchpad.
When to reach for the MessagesState shortcut versus rolling your own schema.
How the builder pattern (add_node / add_edge / .compile()) produces a CompiledStateGraph — and why build_graph() is Forge’s canonical entry point from this module forward.

You’ll build:

ForgeState — Forge’s first typed state schema (issue, messages, repo_path) — plus the forge/ package skeleton and an intake node that summarizes the incoming issue into messages, all wired through build_graph() into a linear START → intake → END graph.

Concepts covered: CA2 — State, schemas & reducers. This is the primary module for state, channels, and reducers; control flow, tools, and persistence arrive in Modules 3, 4, and 6.

Prerequisites: Module 1 completed (repo bootstrapped with uv, config.py / get_model, .env), Python 3.12, uv.

Where we are

✅ M1 — Why LangGraph? Setup, init_chat_model, your first trivial StateGraph, hello Forge and Tally.
👉 M2 — State schemas, channels, reducers, the builder — the birth of ForgeState and the forge/ package.
⬜ M3 — Control flow: conditional edges, cycles, the test-fix loop (Module 3 adds the router and the guard, not today).
⬜ M4–M16 — Tools, parallelism, persistence, HITL, streaming, deployment, evaluation.

Forge before this module: a single throwaway node (hello_graph.py) that can invoke a model, with no package structure and no typed state of its own.

Forge after this module: a forge/ package with a frozen ForgeState (three typed channels) and a skeleton START → intake → END graph that summarizes an incoming issue — ready for the router and the test-fix cycle that Module 3 will bolt on.

State is the contract: TypedDict, Pydantic, or dataclass

In LangGraph, every piece of information your graph needs to carry between nodes flows through a single object called the state schema. It is not a free-form dict that you can add keys to at will — it is a typed contract that every node reads from and writes into. That contract is what makes a StateGraph inspectable, resumable, and checkpointable: the runtime knows exactly what the state looks like at every superstep.

You pass the schema class to StateGraph(ForgeState). Three concrete choices exist:

Form	What it gives you	When to use it
`TypedDict`	Lightweight, no runtime validation, standard Python type hints	Default for Forge — speed, simplicity, no surprise coercions
Pydantic `BaseModel`	Validation on every write, coercion, rich error messages	When inputs arrive from untrusted external sources
`@dataclass`	Attribute-style access, default values without `total=False`	A matter of taste; Forge doesn’t use it

ForgeState is Forge’s typed state, introduced right here in Module 2 and frozen at exactly three channels. It will only ever grow — never be renamed or retyped:

# forge/state.py — the complete file for Module 2
from __future__ import annotations
from typing import Annotated, TypedDict
from langchain_core.messages import AnyMessage
from langgraph.graph.message import add_messages

class ForgeState(TypedDict, total=False):
    # --- Module 2 ---
    issue: str                                           # the bug/feature request
    messages: Annotated[list[AnyMessage], add_messages]  # agent scratchpad
    repo_path: str                                       # path to the sandbox COPY of Tally
    # --- Module 3 adds: route, attempts, test_result ---
    # --- Module 4 adds: plan ---
    # --- Module 5 adds: targets, edits ---
    # --- Module 7 adds: plan_approved, pull_request ---
    # --- Module 12 adds: review ---

total=False lets nodes return partial-state dicts without having to populate every key — standard for LangGraph patterns where nodes only touch the fields they own.

Here is what each field carries at this stage:

issue: str — the raw text of the incoming bug report or feature request. Last-write-wins (no reducer annotation): if a node writes it again, it simply replaces the previous value.
messages: Annotated[list[AnyMessage], add_messages] — the agent’s scratchpad: every question posed, every model response, accumulated in order. The annotation is what makes it accumulate instead of overwrite. More on that in the next section.
repo_path: str — the filesystem path to the sandbox copy of Tally. In Module 2 this is just a field; the actual sandbox copy mechanism (tempdir + shutil.copytree) arrives in Module 3. For now, it is a placeholder in the contract.

The choice of TypedDict over Pydantic is deliberate: Forge’s nodes are called by LangGraph’s Pregel runtime with state dicts it already validates by structure; adding a second layer of runtime validation would slow checkpoints and add noise for no gain in the happy path. If Forge were receiving externally-sourced JSON blobs with unknown structure, Pydantic would be the right call.

Here is how the schema slots into the graph topology:

flowchart LR
    START(["START"]) --> intake["intake\n(node)"]
    intake --> END(["END"])

    subgraph ForgeState
        direction TB
        A["issue: str\n⬤ overwrite"]
        B["messages: list[AnyMessage]\n⬤ add_messages reducer"]
        C["repo_path: str\n⬤ overwrite"]
    end

    intake -.reads.-> A
    intake -.writes to.-> B

The three channels thread through the single node. messages is special: it has a reducer. The other two are plain.

Channels and last-write-wins

Every key in your state schema is a channel. A channel is more than a dict slot: it has a merge semantics that determines what happens when a node (or multiple parallel nodes in the same superstep) writes to it.

The default semantics — what you get with a bare, unannotated field like issue: str — is last-write-wins. If two nodes in the same superstep both write issue, the second write silently overwrites the first. If only one node writes it, it replaces whatever was there before. This is deterministic and cheap, and it is exactly what you want for scalar metadata that a single node owns.

LangGraph ships a sentinel called Overwrite that makes the last-write-wins intent explicit:

from langgraph.types import Overwrite
from typing import Annotated

class ExplicitState(TypedDict):
    # functionally identical to `issue: str`, but documents the intent
    issue: Annotated[str, Overwrite()]

In practice you rarely need Overwrite for simple fields — the bare annotation is the default. It matters when you have a reducer on a channel and need to temporarily bypass it (for example, to reset a running total without going through the accumulation function). Forge does not use it in Module 2, but knowing it exists saves you from confusion later.

The Pregel mental model from Module 1 is what gives last-write-wins its meaning: during a superstep, every node in the active layer runs, queuing its writes. At the end of the superstep, a barrier collects all queued writes and applies them to the state. For overwrite channels, applying a write is simply assignment. For reducer channels, the runtime calls the reducer function instead. The distinction lives entirely in the annotation — or its absence.

Reducers: how two writes merge

The word reducer in LangGraph means the same thing it does in functional programming: a function of the form (current_value, update) -> new_value that knows how to fold a new write into the existing state of a channel.

You attach a reducer to a channel using Annotated:

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

messages: Annotated[list[AnyMessage], add_messages]

That one annotation changes the semantics of the entire messages channel. Instead of overwriting, every write is merged through add_messages.

`add_messages` — the right reducer for the agent scratchpad

add_messages (from langgraph.graph.message) does two things:

Appends new messages to the list if they don’t already exist in it.
Updates by id if a message with the same id already exists — meaning if you retry a node and the model produces a message with the same id, it replaces rather than duplicates it.

That second behavior is what makes LangGraph’s persistence and retry machinery safe: re-executing a node after a crash will not stack up duplicate messages.

The intake node in Forge uses it like this:

# forge/nodes.py — the intake node
from langchain_core.messages import HumanMessage, SystemMessage
from forge import config
from forge.state import ForgeState

INTAKE_SYSTEM = (
    "You are Forge, an autonomous coding agent working on the Tally codebase. "
    "Read the incoming issue and restate it in one or two precise sentences: what "
    "change is requested and which part of the code it likely touches."
)

def intake(state: ForgeState) -> dict:
    """Read the raw issue and write a crisp restatement into the scratchpad."""
    model = config.get_model("fast")
    summary = model.invoke(
        [SystemMessage(content=INTAKE_SYSTEM), HumanMessage(content=state["issue"])]
    )
    return {"messages": [summary]}

Three things to notice:

The node takes ForgeState and returns a partial-state dict — just the keys it touches. It does not touch issue or repo_path. The runtime merges this partial update into the existing state.
The node never mutates state in place. Returning a dict is the contract. Mutating the input dict is undefined behavior in LangGraph.
The response is accessed as a message object; when you later need to display or forward it, reach for .content_blocks — not the bare .content string, which is deprecated in v1.x.

`operator.add` — for lists of homogeneous items

operator.add is Python’s built-in list concatenation, and it works as a reducer for any channel that needs to accumulate a list of items that all have the same shape:

import operator
from typing import Annotated

# Hypothetical: a list of strings grows with every write
notes: Annotated[list[str], operator.add]

In Forge’s full architecture, the edits channel (introduced in Module 5 for the map-reduce fan-in) uses operator.add to collect file edits from multiple parallel Editor nodes. We are not adding that field now — it is mentioned here because it is the conceptual neighbor of add_messages and you will recognize it in Module 5.

One important constraint: a reducer used in fan-out (parallel nodes writing to the same channel in a single superstep) must be associative and commutative — the order in which the parallel writes arrive is not guaranteed. operator.add is commutative for lists only if the order of items does not matter to your logic. add_messages deduplicates by id so it handles re-ordering safely. A custom reducer must be designed with this constraint in mind (Module 5 goes deeper on this).

Custom reducers

Any callable with the signature (current, update) -> new works as a reducer. A practical example — merging two dicts of integer counters without losing counts from either side:

def merge_counters(current: dict, update: dict) -> dict:
    result = dict(current)
    for k, v in update.items():
        result[k] = result.get(k, 0) + v
    return result

# Then in the schema:
# counts: Annotated[dict[str, int], merge_counters]

For Forge in Module 2, the only reducer you need is add_messages. The rest are conceptual neighbors.

`MessagesState` — the shortcut, and when not to use it

LangGraph ships a pre-built state class:

from langgraph.graph import MessagesState

MessagesState is exactly this:

class MessagesState(TypedDict):
    messages: Annotated[list[AnyMessage], add_messages]

One field, one reducer, ready to go. If your agent is a pure conversational agent that only needs a message history, sub-classing MessagesState is the right move:

class MyConversationalState(MessagesState):
    user_id: str  # just add what you need

Forge cannot use it directly because ForgeState needs two more channels — issue and repo_path — that do not belong in a generic conversational schema. Rolling your own TypedDict is the right call when your state is richer than a message list.

⚠️ Common misconception: “I’ll just append to state['messages']” — or there’s a MessageGraph for that.

Both ideas are wrong, and both will haunt you in v1.x.

Misconception 1: “I can accumulate messages by doing state['messages'].append(msg) inside a node.” You can’t — and you shouldn’t try. Nodes receive an immutable view of the state. Mutating it in place is undefined behavior. The correct pattern is to return {"messages": [msg]} from the node and let the add_messages reducer do the accumulation. Without that annotation, writing {"messages": [msg]} replaces the entire list.

Misconception 2: “MessageGraph is the class for graphs that handle messages.” MessageGraph was removed in LangGraph v1.x. It does not exist. Any tutorial or Stack Overflow answer that references MessageGraph is describing v0.x behavior. The v1.x replacement is StateGraph with MessagesState (or your own schema annotated with add_messages). If you see from langgraph.graph import MessageGraph in a codebase, that is a migration target, not a pattern to copy.

The builder and `.compile()`

You now have a schema. Turning it into a runnable graph takes four lines:

# forge/graph.py — the canonical entry point for Module 2
from __future__ import annotations
from langgraph.graph import END, START, StateGraph
from forge import nodes
from forge.state import ForgeState

def build_graph():
    builder = StateGraph(ForgeState)
    builder.add_node("intake", nodes.intake)
    builder.add_edge(START, "intake")
    builder.add_edge("intake", END)
    return builder.compile()

Step by step:

StateGraph(ForgeState) — creates a builder bound to your schema. The graph knows the shape of every state update it will receive.
builder.add_node("intake", nodes.intake) — registers the intake function under the name "intake". The name is what you’ll use in edges and, later, in Command(goto=...).
builder.add_edge(START, "intake") — declares that execution begins at intake. START and END are sentinel constants from langgraph.graph.
builder.add_edge("intake", END) — declares that after intake, the graph terminates.
builder.compile() — produces a CompiledStateGraph: a fully validated, runnable object that implements the LangChain Runnable interface. You get .invoke(), .stream(), .get_state(), and .get_state_history() for free.

build_graph() is Forge’s canonical entry point from this module forward. Every later module extends this function — adding a router in M3, branching the Planner in M4, wiring the Send fan-out in M5. The function signature and name never change.

Calling the compiled graph:

graph = build_graph()
result = graph.invoke({"issue": "tally report crashes on empty expense list"})
# result["messages"] now contains the AI's one-sentence restatement

The input dict only needs the keys that matter to the first node. repo_path is legitimately absent at invocation time in Module 2; total=False in the TypedDict definition makes this legal.

Multiple schemas: input, output, and private state

StateGraph accepts two optional keyword arguments that let you separate what the caller sees from what the graph uses internally:

builder = StateGraph(
    ForgeState,
    input_schema=ForgeInput,    # keys the caller must supply
    output_schema=ForgeOutput,  # keys visible in .invoke() result
)

Any field that is in ForgeState but absent from output_schema becomes private state — it is checkpointed internally but invisible to the caller. This is useful for hiding scratch fields, token counts, or internal routing signals from downstream consumers.

Forge keeps a single schema in Module 2. When Forge grows a Supervisor and subgraphs in Module 12, separating the internal state from the public interface becomes practical. For now, understanding that input_schema= / output_schema= exist (and that the v1.x names are those, not input= / output=) is the right level of preparation.

Hands-on lab: building `ForgeState` from scratch

Goal: Stand up the forge/ package, define ForgeState with its three frozen channels, and confirm with offline tests that messages accumulates and issue overwrites.

At the end of this lab, running uv run python -m pytest tests/ -q prints something like 5 passed in 0.3s — no API key required.

Step 1: the `forge/` package skeleton

# forge/__init__.py
"""Forge — an autonomous coding agent built with LangGraph v1.x.

Module 1 ships only the model layer (`forge.config`). The cumulative graph, state,
tools and nodes arrive from Module 2 onward.
"""

__version__ = "0.1.0"

config.py (introduced in Module 1) is promoted into forge/config.py without any behavioral change. The tier-to-model mapping stays exactly as it was. This is worth saying explicitly: promoting a file into a package is never the moment to refactor it.

Step 2: `forge/state.py`

The real file, from the lab repo:

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

class ForgeState(TypedDict, total=False):
    # --- Module 2 ---
    issue: str
    messages: Annotated[list[AnyMessage], add_messages]
    repo_path: str
    # --- Module 3 adds: route, attempts, test_result ---
    # --- Module 4 adds: plan ---
    # --- Module 5 adds: targets, edits ---
    # --- Module 7 adds: plan_approved, pull_request ---
    # --- Module 12 adds: review ---

The comments are load-bearing documentation: every future module that adds a field knows exactly where to put it. No field is ever renamed, retyped, or removed.

Step 3: `forge/nodes.py`

from __future__ import annotations
from langchain_core.messages import HumanMessage, SystemMessage
from forge import config
from forge.state import ForgeState

INTAKE_SYSTEM = (
    "You are Forge, an autonomous coding agent working on the Tally codebase. "
    "Read the incoming issue and restate it in one or two precise sentences: what "
    "change is requested and which part of the code it likely touches."
)

def intake(state: ForgeState) -> dict:
    """Read the raw issue and write a crisp restatement into the scratchpad."""
    model = config.get_model("fast")
    summary = model.invoke(
        [SystemMessage(content=INTAKE_SYSTEM), HumanMessage(content=state["issue"])]
    )
    return {"messages": [summary]}

The node always reads through config.get_model("fast") — never a bare model id. This means the test suite can monkeypatch forge.config.get_model to inject a fake model and the node runs without a network call.

Step 4: `forge/graph.py`

Already shown above. build_graph() compiles the graph and returns the CompiledStateGraph. No other file in forge/ should call builder.compile() directly — they call build_graph() instead.

Step 5: the offline test suite

The full test harness uses a GenericFakeChatModel that replays canned responses:

# tests/conftest.py — shared fixtures
from langchain_core.language_models.fake_chat_models import GenericFakeChatModel
from langchain_core.messages import AIMessage
import pytest

def fake_model(responses):
    msgs = [AIMessage(content=r) if isinstance(r, str) else r for r in responses]
    return GenericFakeChatModel(messages=iter(msgs))

@pytest.fixture
def make_fake_model():
    return fake_model

And the Module 2 smoke test exercises both behaviors — accumulation and overwrite:

# tests/test_module_02.py
from forge import config, nodes
from forge.state import ForgeState

def test_intake_node_summarizes_issue(monkeypatch, make_fake_model):
    fake = make_fake_model(["Fix the KeyError in tally/report.py for empty categories."])
    monkeypatch.setattr(config, "get_model", lambda tier="fast", **k: fake)
    update = nodes.intake({"issue": "tally report --month crashes on empty category"})
    assert "KeyError" in update["messages"][-1].content

def test_forgestate_introduces_module2_fields():
    keys = set(ForgeState.__annotations__)
    assert {"issue", "messages", "repo_path"} <= keys

Run it:

$ uv run python -m pytest tests/ -q
5 passed in 0.3s

The Tally test suite — real pytest tests against the Tally CLI, no mocking — stays green throughout because Module 2 never touches tally/.

Try it yourself

Exercise 1 — see the reducer in action. Call build_graph().invoke(...) twice in a row, threading the state through. Observe that messages grows. Then temporarily remove the add_messages annotation from state.py, re-run the same calls, and watch the history disappear. Re-add the annotation when done.

Exercise 2 — experiment with operator.add. Add a temporary field notes: Annotated[list[str], operator.add] to your local ForgeState, write a test that issues two partial-state dicts each containing a notes list, and confirm that the results concatenate. Then remove the field — it is not part of the frozen contract.

In production

Keep the state schema lean. Every field in ForgeState gets checkpointed — serialized to the persistence backend — at every superstep boundary. A schema that grows bloated with large blobs (long files, base64 images, redundant copies of data) translates directly into slow checkpoints and expensive storage in Module 6. The rule: if a node only needs a value to complete its own work and no later node reads it, it is a local variable, not a channel.

TypedDict vs Pydantic in the real world. Forge uses TypedDict because the inputs to its nodes come from trusted sources — other nodes in the same graph, or an invocation from your own code. If Forge were exposed via an API and the issue field were populated by arbitrary user HTTP input, a Pydantic model would be the better choice: it validates on write and gives you structured error messages before bad data reaches the LLM.

Reducers in parallel graphs must be associative. In Module 5, multiple Editor nodes write to edits in the same superstep. The order in which their writes land is not guaranteed — the Pregel runtime processes them as they complete. A reducer that assumes a specific write order will produce non-deterministic results. operator.add is safe for unordered accumulation; so is add_messages (deduplication by id). Your own custom reducers need the same property.

Never mutate state in a node. A node that does state["messages"].append(msg) and returns None is incorrect in two ways: it mutates the state object that the runtime is managing, and it returns no partial-state update (so the runtime does not know the write happened). Always return {"messages": [msg]} and let the reducer apply it.

Private state. The output_schema= parameter lets you gate which fields of ForgeState surface in the result of .invoke(). Any field absent from the output schema is private: it travels through the graph, gets checkpointed, but is not visible to the caller. Forge keeps a single unified schema today, but once you add internal fields like attempts (M3) or interim scratch pads, you may want to keep them out of the public API surface. The input_schema= / output_schema= parameters (not input= / output=, which were the v0.x names) give you that control.

Mastery corner

What to really understand here

A reducer is not a convenience — it is the thing that makes multi-node graphs safe. Without a reducer, any two writes to the same channel in the same superstep produce a race. With a reducer, the runtime merges them deterministically. The annotation Annotated[list[AnyMessage], add_messages] is what separates “the scratchpad accumulates” from “the scratchpad gets clobbered.” ForgeState is not a dict you happen to type-hint — it is a contract that every future module extends by addition. Understanding why total=False matters (partial-state returns from nodes) and why TypedDict was chosen over Pydantic (trusted internal inputs, no runtime coercion overhead) is the practical core of this module.

Questions

Q1. You are debugging a Forge graph and notice that after the second node runs, state["messages"] contains only the messages from that second node — all previous messages are gone. What is the most likely cause?

A) The add_messages reducer is deduplicating messages because they share the same id.
B) The messages field is declared as list[AnyMessage] without an Annotated reducer, so last-write-wins semantics apply.
C) The StateGraph automatically clears messages between supersteps to save memory.
D) The node is calling model.invoke() instead of model.stream(), which replaces the message list.

Q2. You are building a new LangGraph agent for processing customer support tickets. Each ticket arrives as JSON from an external API and may have missing or wrongly-typed fields. Which state schema form best fits this scenario?

A) TypedDict — it gives you type hints without overhead, and LangGraph validates inputs automatically.
B) @dataclass — it provides default values, which handle missing fields.
C) A Pydantic BaseModel — it validates and coerces inputs on write, catching malformed data before it reaches the LLM.
D) Any form works; the difference is purely cosmetic.

Q3. You need a reducer that counts how many times each error code has appeared across parallel nodes writing to the same channel in a single superstep. You write:

def count_errors(current: dict, update: dict) -> dict:
    result = dict(current)
    for k, v in update.items():
        result[k] = result.get(k, 0) + v
    return result

Which statement is most accurate?

A) This reducer is correct and safe for parallel writes because it is associative and produces the same result regardless of write order.
B) This reducer is incorrect because LangGraph only accepts operator.add or add_messages as reducers.
C) This reducer is correct but only if parallel writes never share the same key.
D) This reducer is not valid because a custom reducer must return a list, not a dict.

Q4. You are building a pure conversational agent that only needs to track the conversation history. Which is the cleanest approach in LangGraph v1.x?

A) Use MessageGraph from langgraph.graph — it is purpose-built for message accumulation.
B) Sub-class MessagesState (from langgraph.graph) and add your extra fields.
C) Define a TypedDict with messages: list[AnyMessage] (no annotation) and append manually inside each node.
D) Use StateGraph with a Pydantic model so that messages are validated before being stored.

Q5. Forge’s output in production should expose only issue and messages to callers — the internal repo_path field (used only by the sandbox layer) should not appear in the result of .invoke(). Which approach is correct in LangGraph v1.x?

A) Delete repo_path from ForgeState after compilation.
B) Pass output_schema=ForgeOutput (a TypedDict with only issue and messages) to StateGraph(...). Fields absent from the output schema are private and invisible to callers.
C) Pass output=ForgeOutput to StateGraph(...) — the output= keyword controls the public interface.
D) Mark the field with Annotated[str, "private"] — LangGraph recognizes this sentinel and excludes it from output.

Answers

Q1 — B. Without Annotated[list[AnyMessage], add_messages], the messages channel uses last-write-wins semantics. Every node that writes {"messages": [...]} replaces the entire list. The add_messages annotation is not optional when you want accumulation. Option A is plausible but describes normal behavior (deduplication by id), not data loss.

Q2 — C. When inputs are external and untrusted, Pydantic validation on write is the right safeguard. TypedDict has no runtime validation. @dataclass defaults do not validate types. LangGraph itself does not validate field types — the schema describes, it does not enforce.

Q3 — A. The reducer is correct. Integer addition is commutative and associative, so the result is the same regardless of the order parallel writes are applied. Custom reducers are fully supported; they are just callables with the signature (current, update) -> new.

Q4 — B. MessageGraph was removed in LangGraph v1.x. Option C skips the reducer (last-write-wins; the history disappears). Sub-classing MessagesState is the v1.x idiomatic path for pure conversational agents. Pydantic (option D) works but is heavier than necessary for a conversational agent where inputs are trusted.

Q5 — B. The output_schema= parameter (v1.x name; formerly output= in v0.x — do not use that form) accepts a schema class. Fields absent from it are checkpointed internally but invisible to .invoke() callers. Option C uses the deprecated v0.x keyword. Option D invents a sentinel that does not exist.

Traps to avoid

Trap 1 — Reaching for MessageGraph. It does not exist in LangGraph v1.x. The v0.x class was removed. Any resource that references from langgraph.graph import MessageGraph is documenting the past. Use StateGraph with MessagesState or a custom Annotated[list[AnyMessage], add_messages] field. This is one of the highest-frequency v0→v1 migration mistakes.

Trap 2 — Believing “the state is just a dict.” The state is a typed schema with per-channel merge semantics. Without the add_messages reducer annotation, writing {"messages": [new_msg]} from a node replaces the entire message history. The reducer is not optional for accumulation — it is the mechanism. The annotation in Annotated[..., add_messages] is not a type hint; it is a runtime instruction to the Pregel engine.

Trap 3 — Using input= / output= instead of input_schema= / output_schema=. In LangGraph v0.x, StateGraph accepted input= and output= as keyword arguments for multiple schemas. These were renamed to input_schema= and output_schema= in v1.x. Using the old names will either silently be ignored or raise an unexpected keyword argument error depending on your patch version. Always use the v1.x names.

Key takeaways

The state schema is a typed contract — not a free-form dict — that every node in the graph reads from and writes into. You pass it to StateGraph(ForgeState).
Every key in the schema is a channel. A channel without a reducer annotation uses last-write-wins semantics: writing to it replaces whatever was there.
A reducer (Annotated[<type>, <reducer_fn>]) tells the runtime how to merge two writes to the same channel. add_messages accumulates and deduplicates by id; operator.add concatenates; custom reducers work for anything else.
MessagesState is the pre-built shortcut for pure conversational agents. Forge rolls its own ForgeState because it needs issue and repo_path in addition to messages.
The builder pattern — StateGraph(schema) → add_node / add_edge → .compile() — produces a CompiledStateGraph that is a full LangChain Runnable.
build_graph() in forge/graph.py is Forge’s canonical entry point from this module forward. Every module extends it by addition.
input_schema= / output_schema= (v1.x names) let you gate what enters and exits the graph, making any field absent from the output schema private state.

What’s next and where to get it

Want the full LangGraph mastery question bank (interview-style) and the next module in your inbox? Subscribe here — it’s free, like everything in this series.

Forge can hold an issue now — but it still walks one straight line. Module 3 gives it a router (bug vs feature vs chore) and a self-correcting test-fix cycle that loops until the tests go green.

Lab code: github.com/dupuis1212/langgraph-prod-labs/tree/main/module-02
Previous: Module 1 — Why LangGraph? From a While-Loop Agent to a Graph
Next: Module 3 — Control Flow: Conditional Edges, Command, and Self-Correcting Cycles
Full syllabus: LangGraph in Production

References

LangGraph — Concepts: State — schema types, channels, reducers, add_messages. langchain-ai.github.io/langgraph/concepts/low_level
LangGraph — How-to: Define graph state — TypedDict vs Pydantic vs dataclass, MessagesState, multiple schemas. langchain-ai.github.io/langgraph/how-tos/state-model
LangGraph API reference — add_messages (langgraph.graph.message). langchain-ai.github.io/langgraph/reference/graphs/#langgraph.graph.message.add_messages
LangGraph API reference — MessagesState (langgraph.graph). langchain-ai.github.io/langgraph/reference/graphs/#langgraph.graph.MessagesState
Python docs — typing.TypedDict. docs.python.org/3/library/typing.html#typing.TypedDict
Python docs — typing.Annotated. docs.python.org/3/library/typing.html#typing.Annotated