TL;DR: I replaced LangChain's sequential
LLMChainpipelines with LangGraph's state machine architecture for MiningNiti's five-agent document processing system. The result: explicit control flow, retry-on-failure per agent, shared mutable state, and the ability to run agents concurrently with conditional branching. Here's exactly how it works.
When Linear Chains Break
My first instinct when building MiningNiti's AI pipeline was to use LangChain's SequentialChain. Pass a document through Agent 1, take the output, pass it to Agent 2, repeat.
It broke immediately.
Problem 1: Conditional branching. The Compliance Auditor agent needed to run after the Classifier determined a document's type. If the document was a safety protocol, skip the Compliance Auditor entirely. Sequential chains don't branch.
Problem 2: Shared state. All five agents needed access to the original document, the classification result, extracted entities, and the summary. With chains, I was passing growing dictionaries of context through each step — a maintainability nightmare.
Problem 3: Error recovery. If the Entity Extractor failed (rate limit from Cerebras), the entire pipeline died. I needed per-agent retry logic without restarting from scratch.
LangChain's AgentExecutor helps, but it's still fundamentally a loop around a single agent. What I needed was a graph — nodes, edges, conditional routing, and shared state.
The State Machine Model
LangGraph models computation as a directed graph where:
- Nodes are functions (agents, tools, or processors)
- Edges connect nodes (with optional conditions)
- State is a typed dictionary that flows through the graph and is mutated by nodes
- The runtime decides which node to execute next based on the current state
┌──────────────┐
│ START │
└──────┬───────┘
│
▼
┌──────────────┐
│ Classifier │──→ (determines doc type)
└──────┬───────┘
│
┌────────────┼────────────┐
▼ ▼ ▼
┌──────────┐ ┌──────────┐ ┌──────────┐
│ Safety │ │ Entity │ │ Summary │
│ Analyzer │ │ Extractor│ │ Agent │
└────┬─────┘ └────┬─────┘ └────┬─────┘
│ │ │
└─────────────┼────────────┘
│
▼
┌──────────────┐
│ Compliance │ (conditional: only for regulatory docs)
│ Auditor │
└──────┬───────┘
│
▼
┌──────────────┐
│ ASSEMBLE │
│ OUTPUT │
└──────────────┘
The Implementation
Step 1: Define Shared State
HLJS PYTHONfrom typing import TypedDict, Annotated, Literal
from langgraph.graph import add_messages
from operator import add
class MiningState(TypedDict):
"""Shared state flowing through the entire agent graph."""
document_id: str
raw_text: str
doc_type: str # Set by Classifier
classification_confidence: float
safety_hazards: list[str] # Set by Safety Analyzer
entities: dict # Set by Entity Extractor
summary: str # Set by Summarizer
compliance_issues: list[str] # Set by Compliance Auditor
final_output: dict # Assembled result
errors: list[str] # Accumulated errors from any agent
Every node reads from and writes to this state. The runtime handles serialization and concurrency.
Step 2: Define Agent Nodes
Each agent is a function that takes the state and returns a partial update:
HLJS PYTHONimport asyncio
from langchain_groq import ChatGroq
from langchain_mistralai import ChatMistralAI
from langchain_google_genai import ChatGoogleGenerativeAI
async def classifier_agent(state: MiningState) -> dict:
"""Classify document type using Groq (gpt-oss-120b)."""
llm = ChatGroq(model="llama-3.3-70b-versatile", temperature=0)
prompt = f"""Classify this mining document into exactly one category:
- safety_protocol
- environmental_report
- equipment_manual
- regulatory_compliance
- incident_report
Document: {state['raw_text'][:3000]}
Return ONLY a JSON object: {{"type": "...", "confidence": 0.0-1.0}}"""
try:
result = await llm.ainvoke(prompt)
# Parse JSON from response
import json
classification = json.loads(result.content)
return {
"doc_type": classification["type"],
"classification_confidence": classification["confidence"],
}
except Exception as e:
return {"errors": [f"Classifier failed: {str(e)}"], "doc_type": "unknown"}
async def safety_analyzer(state: MiningState) -> dict:
"""Analyze safety hazards using Mistral (Magistral)."""
llm = ChatMistralAI(model="mistral-large-latest", temperature=0)
prompt = f"""Analyze this mining document for safety hazards.
Extract specific hazard mentions with severity levels.
Document type: {state['doc_type']}
Document: {state['raw_text'][:4000]}
Return JSON: {{"hazards": [{"name": "...", "severity": "low|medium|high|critical"}]}}"""
try:
result = await llm.ainvoke(prompt)
import json
data = json.loads(result.content)
return {"safety_hazards": [h["name"] for h in data.get("hazards", [])]}
except Exception as e:
return {"errors": [f"Safety Analyzer failed: {str(e)}"]}
async def entity_extractor(state: MiningState) -> dict:
"""Extract entities using Cerebras (GPT-OSS-120B)."""
from langchain_community.llms import Ollama
llm = ChatGroq(model="llama-3.3-70b-versatile", temperature=0)
prompt = f"""Extract key entities from this mining document:
- Organizations, regulations, equipment, locations, dates, chemical compounds
Document: {state['raw_text'][:4000]}
Return JSON: {{"entities": {{"organizations": [], "regulations": [], ...}}}"""
try:
result = await llm.ainvoke(prompt)
import json
return {"entities": json.loads(result.content)}
except Exception as e:
return {"errors": [f"Entity Extractor failed: {str(e)}"]}
async def summarizer_agent(state: MiningState) -> dict:
"""Generate executive summary using Cerebras."""
from langchain_community.llms import Ollama
llm = ChatGroq(model="llama-3.3-70b-versatile", temperature=0.3)
prompt = f"""Write a concise executive summary of this mining document.
Focus on: key findings, compliance status, and action items.
Document: {state['raw_text'][:5000]}"""
try:
result = await llm.ainvoke(prompt)
return {"summary": result.content}
except Exception as e:
return {"errors": [f"Summarizer failed: {str(e)}"]}
Step 3: Define Conditional Edges
HLJS PYTHONdef should_run_compliance(state: MiningState) -> Literal["compliance_auditor", "assemble_output"]:
"""Only run compliance audit for regulatory documents."""
if state["doc_type"] in ("regulatory_compliance", "safety_protocol"):
return "compliance_auditor"
return "assemble_output"
async def compliance_auditor(state: MiningState) -> dict:
"""Audit document for compliance issues using Gemini."""
llm = ChatGoogleGenerativeAI(model="gemini-2.0-flash", temperature=0)
prompt = f"""Audit this mining document for regulatory compliance issues.
Document type: {state['doc_type']}
Safety hazards found: {state.get('safety_hazards', [])}
Document: {state['raw_text'][:5000]}
Return JSON: {{"issues": [{"regulation": "...", "severity": "...", "description": "..."}]}}"""
try:
result = await llm.ainvoke(prompt)
import json
data = json.loads(result.content)
return {"compliance_issues": [i["description"] for i in data.get("issues", [])]}
except Exception as e:
return {"errors": [f"Compliance Auditor failed: {str(e)}"]}
Step 4: Assemble the Graph
HLJS PYTHONfrom langgraph.graph import StateGraph, START, END
# Build the graph
graph = StateGraph(MiningState)
# Add nodes
graph.add_node("classifier", classifier_agent)
graph.add_node("safety_analyzer", safety_analyzer)
graph.add_node("entity_extractor", entity_extractor)
graph.add_node("summarizer", summarizer_agent)
graph.add_node("compliance_auditor", compliance_auditor)
graph.add_node("assemble_output", lambda state: {
"final_output": {
"doc_type": state["doc_type"],
"safety_hazards": state.get("safety_hazards", []),
"entities": state.get("entities", {}),
"summary": state.get("summary", ""),
"compliance_issues": state.get("compliance_issues", []),
"errors": state.get("errors", []),
}
})
# Define edges
graph.add_edge(START, "classifier")
# After classifier, run 3 agents in parallel
graph.add_edge("classifier", "safety_analyzer")
graph.add_edge("classifier", "entity_extractor")
graph.add_edge("classifier", "summarizer")
# Safety analyzer → conditional compliance audit
graph.add_conditional_edges(
"safety_analyzer",
should_run_compliance,
{"compliance_auditor": "compliance_auditor", "assemble_output": "assemble_output"}
)
# Entity extractor and summarizer → assemble (if compliance not needed)
graph.add_edge("entity_extractor", "assemble_output")
graph.add_edge("summarizer", "assemble_output")
graph.add_edge("compliance_auditor", "assemble_output")
graph.add_edge("assemble_output", END)
# Compile with retry policy
app = graph.compile()
The Gotchas
1. Parallel nodes and state merging
When multiple nodes run concurrently (Safety Analyzer + Entity Extractor + Summarizer), they all write to the same state. LangGraph handles this via Annotated[list, add] — lists are appended, dicts are merged. But if two agents write to the same scalar field, the last writer wins. Design your state to avoid conflicts.
2. Error isolation
Notice each agent catches its own exceptions and returns {"errors": [...]} instead of raising. This means one agent failing doesn't kill the pipeline. The errors list accumulates across all agents, and the final output includes any failures. This is critical for production reliability.
3. Token budget management
Five agents × average 2K tokens per prompt = 10K tokens per document. For a 10K-word document, you're looking at 30K+ tokens total. Truncate aggressively in prompts, and route by token budget rather than by preference: Groq's free tier allows 8K tokens/minute while Cerebras serves the identical gpt-oss-120b at 30K/minute, so the heavier extraction and summarisation work goes to Cerebras and falls back there automatically when Groq rate-limits.
4. Checkpointing
LangGraph supports checkpointing via MemorySaver. For production, use SqliteSaver or PostgresSaver to persist graph state. This enables human-in-the-loop workflows where you can pause, inspect, and resume graph execution.
Results
After migrating from LangChain chains to LangGraph:
- Error isolation: one agent failing no longer kills the run — failures accumulate in
errorsand the rest of the pipeline completes. - Concurrency: the classifier runs first because its category feeds the others; the safety analyzer, entity extractor and summarizer then run together under
asyncio.gather(), and the safety analyzer is skipped outright for non-safety documents. - Cost: $0/month using free-tier providers (Groq + Cerebras + Gemini embeddings).
- Flexibility: adding a new agent = one node + one edge. No pipeline restructuring.
A note on numbers, added later: an earlier version of this post quoted a 98.7% completion rate and a 14s → 3.2s speedup. Those came from ad-hoc runs during the migration and I have not been able to reproduce them under controlled conditions, so I've removed them rather than leave unverifiable figures standing. What is measured, on every CI run, is retrieval quality against a labelled golden set — Hit Rate@5 1.000, MRR 1.000, Recall@5 0.958, nDCG@5 0.968, against floors of 0.90 / 0.75 / 0.85 / 0.75. That gate blocks the build. See the MiningNiti case study for the current architecture.
The Takeaway
-
State machines beat sequential chains for multi-agent systems. Explicit control flow, shared state, and conditional edges give you deterministic behavior that's debuggable and testable. LangGraph's graph model is the right abstraction for complex agent orchestration.
-
Error isolation is non-negotiable. Each agent should catch its own exceptions and write errors to shared state. The pipeline should complete even if individual agents fail. Production systems degrade gracefully — they don't crash.
-
Parallel execution is free with async. Running 3-4 agents concurrently via
asynciocuts pipeline time by 60-70% at zero additional cost. If your agents are I/O-bound (API calls to LLM providers), parallelism is the single highest-ROI optimization you can make.