LangGraph - Building a Parallel Workflow

LangGraph's parallel edge model lets you fan out from a single node to multiple independent branches that execute concurrently, then converge at a single fan-in node. This notebook models an employee onboarding pipeline where HR, Finance, and Infra tasks run in parallel after initiation — rather than sequentially — then merge at a final verification step. The pattern maps directly to any real-world workflow where independent tasks have no data dependency on each other but must all complete before a downstream decision can be made.

Code Snippet:

from langgraph.graph import StateGraph, START, END # Fan-out: all three branches start as soon as "init" completes wf_state.add_edge("init", "infra_p") wf_state.add_edge("init", "hr_p") wf_state.add_edge("init", "finance_p") # Fan-in: verify waits for all three branches to finish wf_state.add_edge("infra_p", "verify") wf_state.add_edge("hr_p", "verify") wf_state.add_edge("finance_p", "verify") def onboarding_verification(state: EmployeeOnBoardingState): # All branch outputs are already merged into state before this runs onboarding_status = "Success" if all([ state["emp_id"], state["joining_kit"], state["email_id"], state["laptop_assigned"], state["bank_acc_linked"], ]) else "Pending" return {"status": onboarding_status}

How it works:

• Fan-out via multiple edges from one node: Adding three add_edge("init", ...) calls tells LangGraph to dispatch all three branches the moment init completes — no manual threading or async coordination needed.

• State acts as the shared accumulator: Each parallel node writes only its own slice (email_id, joining_kit, bank_acc_linked) — because TypedDict fields don't overlap, there are no write conflicts.

• Fan-in is implicit: LangGraph holds the verify node until all incoming edges have resolved. The graph runtime handles synchronization; the node just reads a fully merged state.

• Verification is deterministic: onboarding_verification checks all required fields in one place — it doesn't need to know which branch produced which value.

• Tradeoff: Flat TypedDict state works cleanly here because branches write disjoint keys. If branches wrote the same key, you'd need an Annotated reducer to avoid last-write-wins collisions.

  
  

  Full Code Reference: Parallel Workflow

LangGraph Streaming Modes

LangGraph exposes four streaming modes off a single graph.stream() call, each surfacing a different slice of execution state. Whether you need a full snapshot per node, lightweight deltas, token-by-token LLM output, or raw structured execution logs, there's a mode that fits — and picking the wrong one either floods your UI with redundant data or costs you incremental rendering latency. This post walks through all four using a real intent-classifier routing graph (classify → finance / HR / technical / others) as the running example, so you can see the exact output shape each mode produces.

Code Snippet

from langgraph.graph import MessagesState, StateGraph, START, END from langgraph.checkpoint.memory import InMemorySaver from langchain_core.messages import HumanMessage # Shared setup: graph + checkpointer + thread config checkpointer = InMemorySaver() graph = wf_builder.compile(checkpointer=checkpointer) _config = {"configurable": {"thread_id": "usr-classify-001"}} # ── Mode 1: values ─────────────────────────────────────────────────────────── # Emits the FULL state dict after every node fires. # Shape: complete MsgState snapshot — safe to re-render the whole UI from it. for chunk in graph.stream({"messages": HumanMessage(query)}, config=_config, stream_mode="values"): print(chunk) # {'messages': [...], 'intent': 'technical'} # ── Mode 2: updates ────────────────────────────────────────────────────────── # Emits ONLY the fields each node returned — the delta, not the full state. # Shape: {node_name: partial_state_dict} for chunk in graph.stream({"messages": HumanMessage(query)}, config=_config, stream_mode="updates"): print(chunk) # {'classify_process': {'intent': 'hr'}} # ── Mode 3: messages ───────────────────────────────────────────────────────── # Token-level LLM streaming with per-message metadata. # Shape: (AIMessageChunk, metadata_dict) — unpack .content for incremental text. for chunk, metadata in graph.stream({"messages": HumanMessage(query)}, config=_config, stream_mode="messages"): print(chunk.content, end="") # stream tokens as they arrive # ── Mode 4: debug ──────────────────────────────────────────────────────────── # Structured execution-trace logs — task start/finish events + node metadata. # Shape: raw dict — for observability pipelines, never end-user UI. for log in graph.stream({"messages": HumanMessage(query)}, config=_config, stream_mode="debug"): print(log)

How it works:

Checkpoint + thread_id: InMemorySaver + a stable thread_id in configurable wires multi-turn memory into every graph.stream() call automatically — all four modes respect it.

• values — full snapshot per tick: After each node completes, you receive the entire state object. Ideal when your downstream consumer (UI, logger) is stateless and needs the complete picture to render correctly.
• updates — delta only: Only the fields a node actually mutated come back, keyed by node name. Lower bandwidth for large state objects; requires the consumer to merge deltas into its own copy.
• messages — token streaming: Iterates (AIMessageChunk, metadata_dict) tuples. Buffer tokens into a string and update the display — don't re-render per token or you'll thrash the UI. The metadata dict carries the originating node name, useful for routing multi-node token streams.
• debug — execution trace: Emits task-lifecycle events (task_scheduled, task_result) with full node metadata. Use this for structured logging, latency profiling, or feeding an observability backend — not for user-facing output.

Full Code Reference: Streaming code