Most LangGraph tutorials show toy examples with three nodes and a dictionary state. Enterprise fraud detection requires none of that. After deploying a multi-agent RAG fraud system processing 2,400+ daily alerts across a regulated financial institution, I can tell you exactly what the graph primitives look like under real compliance, latency, and accuracy constraints. This article dissects every node, edge, and state field in our production fraud investigation graph. Every design choice is justified by a specific failure mode we encountered with simpler architectures. Complete code included.
The Complete Graph Topology
Before diving into components, here is the full graph as deployed:

Node count: 9 functional nodes + 1 interrupt point
Edge count: 12 edges (4 conditional)
Cycle count: 1 primary investigation cycle (bounded by depth counter)
State fields: 14 typed fields with explicit reducers
Part 1: The State Object
The state object is the most consequential design decision in any LangGraph system. In chains, state is implicit. In graphs, it is the contract between every node. Get this wrong and every node becomes coupled, untestable, and unauditable.
Production State Definition
from typing import Annotated, TypedDict, Literal, Optional
from langgraph.graph.message import add_messages
from langchain_core.messages import BaseMessage
from pydantic import BaseModel, Field
from enum import Enum
import time
import operator
class RiskTier(str, Enum):
LOW = "low"
MEDIUM = "medium"
HIGH = "high"
CRITICAL = "critical"
class Decision(str, Enum):
APPROVE = "approve"
BLOCK = "block"
ESCALATE = "escalate"
MONITOR = "monitor"
class EvidenceItem(BaseModel):
"""Immutable evidence record. Never mutated after creation."""
source_tool: str
finding_summary: str
confidence_delta: float = Field(ge=-1.0, le=1.0)
timestamp: float = Field(default_factory=time.time)
raw_output: dict = Field(default_factory=dict)
rfc_reference: Optional[str] = None # Regulatory citation for audit
class ToolCallAudit(BaseModel):
"""Complete audit record for every tool invocation."""
call_id: str
tool_name: str
input_params: dict
output_status: Literal["success", "validation_error", "execution_error", "timeout"]
output_summary: str
latency_ms: float
timestamp: float = Field(default_factory=time.time)
def _merge_evidence(existing: list[EvidenceItem], new: list[EvidenceItem]) -> list[EvidenceItem]:
"""Append-only reducer. Evidence is never removed or overwritten."""
return existing + new
def _merge_audit_log(existing: list[ToolCallAudit], new: list[ToolCallAudit]) -> list[ToolCallAudit]:
"""Append-only reducer for audit trail."""
return existing + new
class FraudInvestigationState(TypedDict):
"""
Complete state contract for the fraud investigation graph.
DESIGN PRINCIPLES:
1. Every field has an explicit purpose documented here
2. Mutable collections use append-only reducers
3. Scalar fields are overwritten (last writer wins)
4. No field serves double duty
5. All fields are serializable for checkpointing
"""
# ─── MESSAGE HISTORY ───────────────────────────────────────────
# Full LLM conversation trace including tool calls and results.
# Reducer: add_messages (appends, deduplicates by ID)
messages: Annotated[list[BaseMessage], add_messages]
# ─── CASE IDENTITY (Immutable after INTAKE) ────────────────────
case_id: str # Unique case identifier, e.g., "CASE-20260805-A7F3"
user_id: str # Subject user under investigation
transaction_id: str # Triggering transaction
alert_trigger: str # Rule/model that generated the alert
created_at: float # Case creation timestamp
# ─── INVESTIGATION CONTEXT (Set by TRIAGE, read by all) ────────
risk_tier: Optional[RiskTier] # Initial risk classification
investigation_strategy: Optional[str] # Natural language plan from triage
max_investigation_depth: int # Bounded loop prevention
requires_human_review: bool # Pre-flagged mandatory escalation
# ─── ACCUMULATED EVIDENCE (Append-only) ────────────────────────
# Each item is immutable once added. Confidence is derived, not stored here.
evidence: Annotated[list[EvidenceItem], _merge_evidence]
# ─── DERIVED METRICS (Overwritten each synthesis cycle) ────────
cumulative_confidence: float # Running fraud probability [0.0, 1.0]
investigation_depth: int # Current loop iteration count
# ─── TOOL AUDIT TRAIL (Append-only) ────────────────────────────
tool_audit_log: Annotated[list[ToolCallAudit], _merge_audit_log]
# ─── CONTROL FLOW (Overwritten by routing nodes) ───────────────
current_phase: Literal[
"intake", "triage", "investigation", "synthesis",
"decision", "human_review", "audit_commit"
]
# ─── FINAL OUTPUT (Set once at resolution) ─────────────────────
decision: Optional[Decision]
decision_rationale: Optional[str]
regulatory_citations: list[str] # e.g., ["BSA/AML §1010.320", "FFIEC BSA Manual p.45"]
Why Each Field Exists
| Field | Failure Mode It Prevents |
|---|---|
case_id / created_at | Cross-case state contamination; enables deterministic replay |
investigation_strategy | Agent drift across cycles; keeps investigation focused |
max_investigation_depth | Infinite loops when evidence is inconclusive |
requires_human_review | Compliance-mandated escalations bypassed by confident but wrong models |
evidence (append-only) | Evidence tampering; supports non-repudiation |
cumulative_confidence | Decisions based on single signals instead of accumulated weight |
tool_audit_log | Regulator asks "what did you check?" and you have no answer |
current_phase | Debugging stuck graphs; observability dashboards |
regulatory_citations | Examination findings for missing legal basis in decisions |
Critical: Custom Reducers
The default LangGraph reducer overwrites lists. For fraud evidence and audit logs, this would destroy data. Our custom _merge_evidence and _merge_audit_log reducers enforce append-only semantics. This is not optional—it is a compliance requirement.
Part 2: Node Implementations
Each node is a pure function of state. Nodes do not call each other. They read state, compute, and return partial state updates. The graph orchestrates flow.
Node 1: INTAKE
async def intake_node(state: FraudInvestigationState) -> dict:
"""
Validate alert payload, load historical context from long-term memory,
and initialize case state.
READS: transaction_id, user_id, alert_trigger
WRITES: case_id, created_at, messages (initial context),
requires_human_review (if prior flags exist)
"""
# Load user's historical fraud flags from long-term store
user_history = await store.aget(
namespace=("user_fraud_history", state["user_id"]),
key="flags"
)
prior_flags = user_history.value if user_history else []
requires_human = any(f.get("severity") == "critical" for f in prior_flags)
# Build initial context message for downstream agents
context_msg = HumanMessage(content=(
f"FRAUD ALERT INVESTIGATION\n"
f"Transaction: {state['transaction_id']}\n"
f"User: {state['user_id']}\n"
f"Trigger: {state['alert_trigger']}\n"
f"Prior Flags: {len(prior_flags)}\n"
f"Mandatory Human Review: {requires_human}"
))
return {
"case_id": f"CASE-{datetime.utcnow().strftime('%Y%m%d')}-{uuid.uuid4().hex[:4].upper()}",
"created_at": time.time(),
"messages": [context_msg],
"requires_human_review": requires_human,
"current_phase": "intake",
"cumulative_confidence": 0.0,
"investigation_depth": 0,
"evidence": [],
"tool_audit_log": [],
"regulatory_citations": [],
}Node 2: TRIAGE
TRIAGE_SYSTEM_PROMPT = """You are a fraud triage specialist. Given an alert, classify risk
and define an investigation strategy. Respond in JSON:
{
"risk_tier": "low|medium|high|critical",
"strategy": "Natural language investigation plan",
"max_depth": 3-10 based on risk tier,
"priority_tools": ["ordered list of first tools to call"]
}"""
async def triage_node(state: FraudInvestigationState) -> dict:
"""
Classify risk tier and set investigation parameters.
READS: messages (intake context), requires_human_review
WRITES: risk_tier, investigation_strategy, max_investigation_depth, messages
"""
response = await triage_llm.ainvoke([
SystemMessage(content=TRIAGE_SYSTEM_PROMPT),
*state["messages"]
])
parsed = json.loads(response.content)
# Override max_depth if human review is mandated
max_depth = parsed["max_depth"]
if state["requires_human_review"]:
max_depth = min(max_depth, 3) # Limit auto-investigation before escalation
return {
"risk_tier": RiskTier(parsed["risk_tier"]),
"investigation_strategy": parsed["strategy"],
"max_investigation_depth": max_depth,
"messages": [AIMessage(content=f"Triage complete. Strategy: {parsed['strategy']}")],
"current_phase": "triage",
}
Node 3: INVESTIGATE AGENT (Core Reasoning Node)
INVESTIGATION_PROMPT_TEMPLATE = """You are investigating potential synthetic identity fraud.
CASE STATE:
- Risk Tier: {risk_tier}
- Strategy: {strategy}
- Evidence Collected: {evidence_count} items
- Cumulative Confidence: {confidence:.3f}
- Depth: {depth}/{max_depth}
- Prior Tools Called: {tools_called}
AVAILABLE TOOLS: lookup_transaction, get_device_fingerprint, screen_sanctions,
query_behavioral_biometrics, check_kyc_documents, search_external_watchlists
RULES:
1. Follow the investigation strategy unless evidence contradicts it.
2. Never repeat a tool+params combination already in tool_audit_log.
3. If confidence >= 0.85 with >= 3 evidence items, stop investigating.
4. If depth >= max_depth, stop investigating.
5. Select AT MOST 2 tools per iteration to maintain focus.
6. Explain your reasoning BEFORE calling tools."""
async def investigate_node(state: FraudInvestigationState) -> dict:
"""
Core reasoning node. Decides which tools to call next based on
accumulated evidence and investigation strategy.
READS: messages, risk_tier, investigation_strategy, evidence,
cumulative_confidence, investigation_depth, tool_audit_log
WRITES: messages (with tool_calls), current_phase
"""
tools_called = [a.tool_name for a in state["tool_audit_log"]]
prompt = INVESTIGATION_PROMPT_TEMPLATE.format(
risk_tier=state["risk_tier"],
strategy=state["investigation_strategy"],
evidence_count=len(state["evidence"]),
confidence=state["cumulative_confidence"],
depth=state["investigation_depth"],
max_depth=state["max_investigation_depth"],
tools_called=tools_called
)
response = await investigation_llm.ainvoke([
SystemMessage(content=prompt),
*state["messages"]
])
return {
"messages": [response],
"current_phase": "investigation",
}Node 4: TOOL EXECUTOR (Validated Execution + Audit)
async def tool_executor_node(state: FraudInvestigationState) -> dict:
"""
Execute tool calls with validation, timing, and audit logging.
Returns structured errors instead of raising exceptions.
READS: messages (last AI message with tool_calls)
WRITES: messages (tool results), tool_audit_log, current_phase
"""
last_ai = state["messages"][-1]
if not hasattr(last_ai, "tool_calls") or not last_ai.tool_calls:
return {"current_phase": "synthesis"}
audit_records = []
tool_messages = []
for tc in last_ai.tool_calls:
start = time.monotonic()
try:
result = await execute_validated_tool(tc["name"], tc["args"])
status = result.get("status", "success")
summary = str(result)[:500]
except Exception as e:
result = {"status": "execution_error", "message": str(e)[:200]}
status = "execution_error"
summary = str(e)[:500]
latency = (time.monotonic() - start) * 1000
audit_records.append(ToolCallAudit(
call_id=tc["id"],
tool_name=tc["name"],
input_params=tc["args"],
output_status=status,
output_summary=summary,
latency_ms=round(latency, 2)
))
tool_messages.append(ToolMessage(
content=json.dumps(result),
tool_call_id=tc["id"]
))
return {
"messages": tool_messages,
"tool_audit_log": audit_records,
"current_phase": "synthesis",
}Node 5: EVIDENCE SYNTHESIZER + CONFIDENCE SCORER
# Domain-specific confidence scoring rules
CONFIDENCE_RULES = {
"lookup_transaction": lambda r: 0.25 if r.get("linked_accounts", []) and
len(r["linked_accounts"]) > 5 else 0.10,
"get_device_fingerprint": lambda r: 0.30 if r.get("fingerprint", {}).get("geo_mismatch") else 0.05,
"screen_sanctions": lambda r: 0.35 if r.get("matches_found", 0) > 0 else 0.0,
"check_kyc_documents": lambda r: 0.40 if r.get("forgery_detected") else 0.0,
"query_behavioral_biometrics": lambda r: 0.20 if r.get("anomaly_score", 0) > 0.8 else 0.05,
}
async def synthesize_node(state: FraudInvestigationState) -> dict:
"""
Process tool results into structured evidence and update confidence.
This is where raw tool output becomes auditable findings.
READS: messages (tool results), tool_audit_log, evidence, cumulative_confidence
WRITES: evidence, cumulative_confidence, investigation_depth, messages, current_phase
"""
new_evidence = []
total_delta = 0.0
# Match tool results to audit records
recent_audits = [a for a in state["tool_audit_log"]
if a.output_status == "success"]
for audit in recent_audits[-2:]: # Process only newest tool calls
scoring_fn = CONFIDENCE_RULES.get(audit.tool_name)
if not scoring_fn:
continue
# Re-parse tool output for scoring
result_msg = next(
(m for m in state["messages"]
if getattr(m, "tool_call_id", None) == audit.call_id),
None
)
if not result_msg:
continue
result = json.loads(result_msg.content)
delta = scoring_fn(result)
if delta > 0:
new_evidence.append(EvidenceItem(
source_tool=audit.tool_name,
finding_summary=audit.output_summary[:200],
confidence_delta=delta,
raw_output=result
))
total_delta += delta
updated_confidence = min(1.0, state["cumulative_confidence"] + total_delta)
return {
"evidence": new_evidence, # Appended via custom reducer
"cumulative_confidence": updated_confidence,
"investigation_depth": state["investigation_depth"] + 1,
"messages": [AIMessage(content=(
f"Synthesis: {len(new_evidence)} new evidence items. "
f"Confidence: {updated_confidence:.3f}. "
f"Depth: {state['investigation_depth'] + 1}/{state['max_investigation_depth']}"
))],
"current_phase": "synthesis",
}Node 6: DECISION GATE (Conditional Router)
def decision_gate_router(state: FraudInvestigationState) -> Literal["block_auto", "approve_auto", "escalate"]:
"""
Pure function determining next phase based on accumulated state.
NO LLM CALL. Deterministic logic only for auditability.
"""
conf = state["cumulative_confidence"]
evidence_count = len(state["evidence"])
depth = state["investigation_depth"]
max_depth = state["max_investigation_depth"]
mandatory_human = state["requires_human_review"]
# Mandatory escalation overrides everything
if mandatory_human:
return "escalate"
# Sufficient confidence + evidence → automated decision
if conf >= 0.85 and evidence_count >= 3:
return "block_auto"
# Exhausted investigation without sufficient evidence → approve
if depth >= max_depth and conf < 0.5:
return "approve_auto"
# Exhausted investigation with moderate confidence → escalate
if depth >= max_depth:
return "escalate"
# Should not reach here; safety net
return "escalate"
async def decision_node(state: FraudInvestigationState) -> dict:
"""Generate decision rationale using LLM, citing specific evidence."""
evidence_summary = "\n".join(
f"- [{e.source_tool}] {e.finding_summary} (Δ={e.confidence_delta:.2f})"
for e in state["evidence"]
)
response = await decision_llm.ainvoke([
SystemMessage(content=(
"Generate a regulatory-compliant decision rationale. "
"Cite specific evidence items. Reference applicable regulations."
)),
HumanMessage(content=(
f"Decision: {'BLOCK' if state['cumulative_confidence'] >= 0.85 else 'APPROVE'}\n"
f"Confidence: {state['cumulative_confidence']:.3f}\n"
f"Evidence:\n{evidence_summary}"
))
])
decision = Decision.BLOCK if state["cumulative_confidence"] >= 0.85 else Decision.APPROVE
return {
"decision": decision,
"decision_rationale": response.content,
"messages": [response],
"current_phase": "decision",
}Node 7: HUMAN REVIEW (Interrupt Point)
async def human_review_node(state: FraudInvestigationState) -> dict:
"""
Pauses execution. State is checkpointed automatically.
Resumed via app.ainvoke(None, config=config, command={"resume": {...}})
"""
# This node's primary purpose is the INTERRUPT, not computation.
# When resumed, the analyst's decision is injected via command.
return {
"current_phase": "human_review",
"messages": [HumanMessage(content="Awaiting analyst review...")]
}Part 3: Edge Definitions
Edges are where the graph's intelligence lives. Four types used:
Normal Edges (Unconditional)
graph.add_edge(START, "intake")
graph.add_edge("intake", "triage")
graph.add_edge("triage", "investigate")
graph.add_edge("investigate", "tool_executor")
graph.add_edge("tool_executor", "synthesize")
graph.add_edge("block_auto", "audit_commit")
graph.add_edge("approve_auto", "audit_commit")
graph.add_edge("escalate", "human_review")
graph.add_edge("human_review", "audit_commit") # After resume
graph.add_edge("audit_commit", END)Conditional Edge: Investigation Cycle
def should_continue_investigation(state: FraudInvestigationState) -> Literal["continue", "decide"]:
"""Bounded cycle control. NO LLM. Pure state inspection."""
if state["investigation_depth"] >= state["max_investigation_depth"]:
return "decide"
if state["cumulative_confidence"] >= 0.85 and len(state["evidence"]) >= 3:
return "decide"
if state["requires_human_review"]:
return "decide"
return "continue"
graph.add_conditional_edges(
"synthesize",
should_continue_investigation,
{"continue": "investigate", "decide": "decision_gate"}
)Conditional Edge: Decision Routing
graph.add_conditional_edges(
"decision_gate",
decision_gate_router,
{
"block_auto": "block_auto",
"approve_auto": "approve_auto",
"escalate": "escalate"
}
)Part 4: Compilation and Execution
from langgraph.graph import StateGraph, START, END
graph = StateGraph(FraudInvestigationState)
# Register all nodes
graph.add_node("intake", intake_node)
graph.add_node("triage", triage_node)
graph.add_node("investigate", investigate_node)
graph.add_node("tool_executor", tool_executor_node)
graph.add_node("synthesize", synthesize_node)
graph.add_node("decision_gate", lambda state: {}) # Pure router, no computation
graph.add_node("block_auto", decision_node)
graph.add_node("approve_auto", decision_node)
graph.add_node("escalate", lambda state: {"decision": Decision.ESCALATE})
graph.add_node("human_review", human_review_node)
graph.add_node("audit_commit", audit_commit_node)
# Register all edges (as defined above)
# ... (see Parts 3a-3d)
app = graph.compile(
checkpointer=AsyncPostgresSaver.from_conn_string(DB_URL),
store=AsyncPostgresStore.from_conn_string(DB_URL, index={"dims": 1536}),
interrupt_before=["human_review"]
)Running a Case
config = {
"configurable": {
"thread_id": "CASE-20260805-A7F3",
"user_id": "fraud_ops_team",
"session_id": "shift_20260805_am"
}
}
result = await app.ainvoke({
"messages": [],
"user_id": "USR-99281",
"transaction_id": "TX-K8M2P4Q7",
"alert_trigger": "velocity_rule_v3",
}, config=config)
# If interrupted for human review:
await app.ainvoke(None, config=config, command={
"resume": {"analyst_decision": "block", "notes": "Confirmed synthetic SSN pattern"}
})Validation: How We Know This Works
Structural Tests
def test_graph_structure():
"""Verify graph topology matches compliance requirements."""
assert len(graph.nodes) == 11
assert graph.get_edge("synthesize", "investigate") is not None # Cycle exists
assert "human_review" in app.interrupt_before
assert all(hasattr(FraudInvestigationState.__annotations__[k], '__metadata__')
or k in ("messages",) # Messages uses built-in reducer
for k in FraudInvestigationState.__annotations__)State Integrity Tests
async def test_evidence_append_only():
"""Evidence must never be lost or overwritten."""
state = {"evidence": [EvidenceItem(source_tool="a", finding_summary="test", confidence_delta=0.1)]}
update = {"evidence": [EvidenceItem(source_tool="b", finding_summary="test2", confidence_delta=0.2)]}
merged = _merge_evidence(state["evidence"], update["evidence"])
assert len(merged) == 2
assert merged[0].source_tool == "a"
assert merged[1].source_tool == "b"End-to-End Replay Test
async def test_deterministic_replay():
"""Same inputs + same checkpoints = identical outputs."""
config = {"configurable": {"thread_id": "replay-test"}}
result1 = await app.ainvoke(initial_state, config=config)
result2 = await app.ainvoke(None, config=config) # Replay from checkpoint
assert result1["decision"] == result2["decision"]
assert len(result1["evidence"]) == len(result2["evidence"])Summary: The Three Primitives
| Primitive | Count | Key Design Constraint |
|---|---|---|
| Nodes | 9 functional + 1 interrupt | Pure functions of state; no inter-node calls |
| Edges | 8 normal + 4 conditional | Cycles bounded by counters; conditionals are deterministic |
| State Fields | 14 typed fields | Append-only reducers for collections; explicit purpose for every field |
The graph is not a diagram. It is a compliance artifact, a debugging interface, and a runtime engine simultaneously. Every node, edge, and state field exists because a simpler alternative failed in production. The complexity is not accidental it is the minimum necessary to operate a fraud detection system that regulators will accept and customers will trust.

Join the conversation! Your thoughts help the community grow.