You already have an n8n workflow that functions.
It receives a request, calls APIs, uses an LLM, makes decisions, updates a database, and returns a result.
During the prototype stage, this is often enough.
But as the workflow grows, things can start becoming harder to manage.
These are scenarios where moving the agent logic from an n8n prototype into LangGraph becomes purposeful.
Replacing n8n just because LangGraph is newer is not quite the point.
The goal is to move the parts that require stateful agent orchestration, explicit control flow, persistence, and long-running execution into a framework designed for those problems.
This guide walks through that migration step by step.
Consider a real estate AI assistant that receives a buyer request:
User Request
↓
Extract Requirements
↓
Search CRM
↓
Assign to AI Agent
↓
Call Property Search API
↓
Score Results
↓
Send Response
↓
Update CRM
Consequently, an n8n workflow might contain:
This is a perfectly reasonable architecture for a prototype.
However, the problem starts becoming noticeable when the workflow becomes something like:
Webhook
↓
20+ nodes
↓
Multiple IF branches
↓
AI Agent
↓
Multiple tool calls
↓
Retries
↓
Human approval
↓
CRM update
↓
Follow-up
↓
Scheduled continuation
At this point, the workflow is doing more than simple automation.
It is becoming an agentic state machine.
This is the perfect stage to evaluate whether you should move the core agent logic into LangGraph.
Don't start rewriting the entire n8n workflow immediately.
First, inspect every node and determine what responsibility it actually performs.
A useful mapping looks like this:
| n8n Component | Responsibility | LangGraph Equivalent |
|---|---|---|
| Webhook | Receive input | API layer |
| Set/Edit Fields | Transform data | Python function |
| Code | Business logic | Python function |
| IF/Switch | Routing | Conditional edge |
| AI Agent | Reasoning | Agent/LLM node |
| HTTP Request | External operation | Tool |
| Database | Data persistence | DB/service |
| Wait | Long-running state | Persistence/interrupt |
| Human approval | Manual decision | interrupt() |
| Error workflow | Recovery | Retry/recovery logic |
This classification prevents one of the biggest migration mistakes:
Rewriting the entire system when only the agent orchestration needs to change.
The CRM, property database, or your external APIs don't necessarily need to move.
The migration should focus on the orchestration layer.
Before converting anything, define exactly what the existing workflow does.
For example:
Buyer Request
↓
Extract Requirements
↓
Search Properties
↓
Filter Results
↓
AI Ranker
↓
Return Recommendations
A buyer might send:
Looking for a 3-bedroom apartment in Dubai Marina
under AED 2 million.
The workflow needs to:
This gives us a clear baseline for the migration.
This is one of the most important changes during the migration.
In an n8n workflow, execution data naturally flows from one node to another.
With LangGraph, you explicitly define the state shared across the graph.
from typing import TypedDict
class AgentState(TypedDict):
user_query: str
requirements: dict
candidates: list
matches: list
selected_property: dict | None
error: str | None
Now the agent has an explicit state contract.
The workflow can move through:
user_query
↓
requirements
↓
candidates
↓
matches
↓
selected_property
This makes the state easier to inspect, test, persist, and reason about.
A useful rule is:
If a piece of information is required by multiple stages of the agent, consider making it part of the graph state.
The next step is to convert individual workflow operations into graph nodes.
The original n8n flow:
Webhook
↓
Code
↓
HTTP Request
↓
AI Agent
↓
IF
Can become:
START
↓
normalize_request
↓
search_properties
↓
rank_properties
↓
route_result
A basic graph can be created like this:
from langgraph.graph import StateGraph, START, END
builder = StateGraph(AgentState)
builder.add_node("normalize_request", normalize_request)
builder.add_node("search_properties", search_properties)
builder.add_node("rank_properties", rank_properties)
builder.add_edge(START, "normalize_request")
builder.add_edge("normalize_request", "search_properties")
builder.add_edge("search_properties", "rank_properties")
builder.add_edge("rank_properties", END)
graph = builder.compile()
The important architectural difference is that the workflow is now represented explicitly as a graph.
Each node has a defined responsibility.
So far, we have identified the nodes' functionalities, and they are mapped to LangGraph nodes. Now, we move the tools.
An n8n HTTP Request node might currently call a property API.
So, instead of letting the agent directly deal with raw HTTP logic, wrap the operation as a tool.
from langchain_core.tools import tool
@tool
def search_properties(
location: str,
max_budget: int,
bedrooms: int
):
"""
Search available properties.
"""
results = property_service.search(
location=location,
max_budget=max_budget,
bedrooms=bedrooms
)
return results
The tool should have:
The important distinction is:
The tool operates. The agent decides when to use it.
This keeps the agent's reasoning separate from infrastructure code.
This is another essential migration step.
Suppose the n8n workflow contains:
IF match_score > 0.8
↓
Strong Match
In LangGraph, make that routing explicit.
def route_match(state: AgentState):
if not state["matches"]:
return "no_match"
if state["matches"][0]["score"] >= 0.8:
return "strong_match"
return "weak_match"
Then, connect the routes with:
builderadd_conditional_edges(
"rank_properties",
route_match,
{
"strong_match": "send_recommendation",
"weak_match": "request_more_preferences",
"no_match": "fallback_search"
}
)
The resulting graph then becomes:
rank_properties
↓
route_match
/ | \
/ | \
strong_match weak_match no_match
↓ ↓ ↓
recommendation ask user fallback search
This is much easier to reason when the number of branches increases.
A prototype often relies on execution history.
A production agent cannot assume that the entire execution will always remain active.
So, consider:
Agent starts
↓
Search properties
↓
Human approval required
↓
Wait 6 hours
↓
Continue
The agent needs to remember where it was and what state it had and here is where LangGraph persistence becomes important.
Conceptually:
Agent State
↓
Checkpoint
↓
Thread
↓
Resume Execution
Compile the graph with a checkpointer:
graph = builder.compile(
checkpointer=checkpointer
)
Then invoke it with a stable thread ID:
config = {
"configurable": {
"thread_id": "lead-123"
}
}
result = graph.invoke(
initial_state,
config=config
)
The thread_id gives the execution a durable identity.
This becomes particularly important for:
interrupt()
Consider an n8n workflow:
AI recommends property
↓
Wait
↓
Agent approval
↓
Continue
A LangGraph implementation can model the same process using an interrupt:
AI Recommendation
↓
interrupt()
↓
Human Decision
↓
Resume Graph
python
from langgraph.types import interrupt
def approval_node(state):
decision = interrupt({
"message": "Approve this property recommendation?",
"property": state["selected_property"]
})
return {
"approval": decision
}
The important difference is that the agent doesn't need to remain continuously active while waiting.
The state can be persisted and execution can resume when the human decision arrives.
This is especially useful for workflows involving:
This is one of the most significant production changes.
Imagine the agent executes:
send_whatsapp_message()
Then, the process crashes immediately afterward.
Next, when the graph resumes, the operation might run again.
You could end up with:
Message sent
↓
Process crashes
↓
Graph resumes
↓
Message sent again
The result is a duplicate customer message.
Instead, design external side effects to be idempotent.
Agent Decision
↓
Generate operation_id
↓
Check idempotency store
↓
Execute side effect
↓
Persist result
A simple implementation might use:
def send_message_once(operation_id, message):
if already_processed(operation_id):
return get_previous_result(operation_id)
result = send_message(message)
save_result(
operation_id=operation_id,
result=result
)
return result
This pattern is especially important for:
A production agent should always assume that execution may be retried or resumed.
Don't rely on the LLM to decide:
"If the API fails, try again."
Retry behavior belongs in the application layer.
from tenacity import retry
from tenacity import stop_after_attempt
from tenacity import wait_exponential
@retry(
stop=stop_after_attempt(3),
wait=wait_exponential()
)
def call_property_api():
return property_api.search()
But not every error should be retried.
Usually:
The workflow should distinguish between these distinctive cases.
API Error
↓
Retryable?
/ \
Yes No
↓ ↓
Retry Recovery
This keeps reliability logic deterministic.
Avoid returning vague errors like:
raise Exception("API failed")
Instead, return structured information.
{
"success": False,
"error_type": "rate_limit",
"retryable": True,
"message": "Property API rate limit exceeded"
}
Now the application can make a deterministic decision.
Tool Result
↓
success?
/ \
Yes No
↓ ↓
Continue retryable?
/ \
Yes No
↓ ↓
Retry Recovery
This is much safer than expecting an LLM to interpret every infrastructure failure that may be possible.
Another easy mistake during migration is putting everything inside the LLM.
Don't do that.
The architecture should, therefore, look like:
Agent
↓
┌────────┴─────────┐
↓ ↓
Deterministic Logic AI Reasoning
↓ ↓
Database / APIs LLM + Tools
The LLM should not be responsible for decisions that can be reliably enforced in code.
A successful demo is not the same thing as a production-ready agent.
Make a rule to test individual nodes first.
def test_route_strong_match():
state = {
"matches": [
{"score": 0.91}
]
}
assert route_match(state) == "strong_match"
Test tools separately:
def test_property_search():
result = search_properties.invoke({
"location": "Dubai Marina",
"max_budget": 2000000,
"bedrooms": 3
})
ascertain that the result is not None
Then test the failure scenarios.
Also, remember to test whether the same execution can safely resume.
Don't only test the final response but also test what happens between nodes.
Input
↓
Normalization
↓
Search
↓
Ranking
↓
Routing
↓
Recommendation
For each transition, verify:
This makes debugging much easier than testing the agent only from the outside.
Also Explore our n8n Workflow Automation service → https://ciphernutz.com/service/n8n-workflow-automation
A common mistake is treating the migration as simple as:
n8n → LangGraph
and having nothing left in n8n.
That isn't always necessary.
A better architecture can be:
n8n
│
┌──────────┼──────────┐
↓ ↓ ↓
Webhooks CRM Events Scheduled Jobs
│
↓
LangGraph
│
┌──────────┼──────────┐
↓ ↓ ↓
State Tools Reasoning
│ │ │
└──────────┼──────────┘
↓
Agent Result
↓
n8n
↓
Notifications / CRM
n8n can continue handling:
Similarly, LangGraph can handle:
This creates a hybrid architecture rather than forcing everything into one platform.
A production architecture could look like this:
┌───────────────┐
│ API / Webhook │
└───────┬───────┘
↓
┌─────────────────┐
│ LangGraph Agent │
└────────┬────────┘
↓
┌────────────────┐
│ Agent State │
└────────┬───────┘
↓
┌────────────────────────────────┐
│ │
Deterministic Agentic
Nodes Nodes
│ │
↓ ↓
Database / APIs LLM + Tools
│ │
└──────────────┬─────────────────┘
↓
Checkpointer
↓
Pos