必收藏！LangGraph与n8n/Dify低代码平台，构建AI Agent保姆级对比（小白&程序员必看）
对于正在迷茫择业、想转行提升，或是刚入门的程序员、编程小白来说，有一个问题几乎人人都在问：未来10年，什么领域的职业发展潜力最大？答案只有一个：人工智能（尤其是大模型方向）当下，人工智能行业正处于爆发式增长期，其中大模型相关岗位更是供不应求，薪资待遇直接拉满——字节跳动作为AI领域的头部玩家，给硕士毕业的优质AI人才（含大模型相关方向）开出的月基础工资高达5万—6万元；即便是非“人才计划”的普通应
本文将系统拆解LangGraph底层编排框架与n8n、Dify等可视化低代码平台，在AI Agent构建中的核心差异、适用场景，帮小白快速分清选型方向，助程序员高效落地项目。LangGraph主打精细化控制，适配复杂生产级智能系统，自带状态管理、检查点、人工中断等高级特性；低代码平台则靠拖拽式可视化界面降低开发门槛，适合快速搭建原型、落地简单业务场景。重点强调：低代码是AI Agent探索的起点，绝非生产部署的终点，核心业务系统仍需依托代码级开发（如LangGraph），才能保障系统可靠性、可维护性，避免后期陷入迭代困境。
LangGraph
LangGraph is a low-level orchestration framework and runtime for building, managing, and deploying long-running, stateful agents。
LangGraph中最重要的两个概念是Node和State，Node通过edge链接，形成一个可执行的workflow，State是整个workflow的Context（上下文），这是一个典型的[Procedure Context]上下文设计模式。基本所有的框架都会使用使用上下文模式，因为处理信息需要上下文，而且这些信息需要在不同Node之间传递。
接下来，我们以一个自动处理用户邮件的AI Agent为例，来演示LangGraph的使用和主要概念，需求是要借助AI的能力智能处理用户的email，其主要处理流程如下：
- 阅读收到的客户邮件
- 根据紧急程度和主题进行分类
- 搜索相关文档以回答问题
- 起草适当的回复
- 将复杂问题升级给人工客服处理
- 处理完成后，存档用户请求
基于LangGraph构建应用，首先我们要对问题进行结构化分解，把每一个处理单元作为一个节点，同时想清楚需要在不同Node之间共享数据的State。
于以上分析，本应用的功能实现代码如下：
import functools
from typing import TypedDict, Literal
import psycopg
from langgraph.checkpoint.postgres import PostgresSaver
from langgraph.store.postgres import PostgresStore
from mermaid_image import generate_mermaid_image_advanced
# Define the structure for email classification
# This is using output format for Classification purpose
class EmailClassification(TypedDict):
intent: Literal["question", "bug", "billing", "feature", "complex"]
urgency: Literal["low", "medium", "high", "critical"]
topic: str
summary: str
class EmailAgentState(TypedDict):
# Raw email data
email_content: str
sender_email: str
email_id: str
# Classification result
classification: EmailClassification | None
# Raw search/API results
search_results: list[str] | None # List of raw document chunks
customer_history: dict | None # Raw customer data from CRM
# Generated content
draft_response: str | None
messages: list[str] | None
from typing import Literal
from langgraph.graph import StateGraph, START, END
from langgraph.types import interrupt, Command
from langchain.messages import HumanMessage
from deepseek_model import llm
def read_email(state: EmailAgentState) -> dict:
"""Extract and parse email content"""
# In production, this would connect to your email service
return {
"messages": [HumanMessage(content=f"Processing email: {state['email_content']}")]
}
def classify_intent(state: EmailAgentState) -> Command[
Literal["search_documentation", "human_review", "bug_tracking"]]:
"""Use LLM to classify email intent and urgency, then route accordingly"""
# Create structured LLM that returns EmailClassification dict
structured_llm = llm.with_structured_output(EmailClassification)
# Format the prompt on-demand, not stored in state
classification_prompt = f"""
Analyze this customer email and classify it:
Email: {state['email_content']}
From: {state['sender_email']}
Provide classification including intent, urgency, topic, and summary.
"""
# Get structured response directly as dict
classification = structured_llm.invoke(classification_prompt)
# Determine next node based on classification
if classification['intent'] == 'billing' or classification['urgency'] == 'critical':
goto = "human_review"
elif classification['intent'] in ['question', 'feature']:
goto = "search_documentation"
elif classification['intent'] == 'bug':
goto = "bug_tracking"
else:
goto = "draft_response"
# Store classification as a single dict in state
return Command(
update={"classification": classification},
goto=goto
)
def search_documentation(state: EmailAgentState) -> Command[Literal["draft_response"]]:
"""Search knowledge base for relevant information"""
# Build search query from classification
classification = state.get('classification', {})
query = f"{classification.get('intent', '')} {classification.get('topic', '')}"
try:
# Implement your search logic here
# Store raw search results, not formatted text
search_results = [
"Reset password via Settings > Security > Change Password",
"Password must be at least 12 characters",
"Include uppercase, lowercase, numbers, and symbols"
]
except Exception as e:
# For recoverable search errors, store error and continue
search_results = [f"Search temporarily unavailable: {str(e)}"]
return Command(
update={"search_results": search_results}, # Store raw results or error
goto="draft_response"
)
def bug_tracking(state: EmailAgentState) -> Command[Literal["draft_response"]]:
"""Create or update bug tracking ticket"""
# Create ticket in your bug tracking system
ticket_id = "BUG-12345" # Would be created via API
return Command(
update={
"search_results": [f"Bug ticket {ticket_id} created"],
"current_step": "bug_tracked"
},
goto="draft_response"
)
def draft_response(state: EmailAgentState) -> Command[Literal["human_review", "send_reply"]]:
"""Generate response using context and route based on quality"""
classification = state.get('classification', {})
# Format context from raw state data on-demand
context_sections = []
if state.get('search_results'):
# Format search results for the prompt
formatted_docs = "\n".join([f"- {doc}" for doc in state['search_results']])
context_sections.append(f"Relevant documentation:\n{formatted_docs}")
if state.get('customer_history'):
# Format customer data for the prompt
context_sections.append(f"Customer tier: {state['customer_history'].get('tier', 'standard')}")
# Build the prompt with formatted context
draft_prompt = f"""
Draft a response to this customer email:
{state['email_content']}
Email intent: {classification.get('intent', 'unknown')}
Urgency level: {classification.get('urgency', 'medium')}
{chr(10).join(context_sections)}
Guidelines:
- Be professional and helpful
- Address their specific concern
- Use the provided documentation when relevant
"""
response = llm.invoke(draft_prompt)
# Determine if human review needed based on urgency and intent
needs_review = (
classification.get('urgency') in ['high', 'critical'] or
classification.get('intent') == 'complex'
)
# Route to appropriate next node
goto = "human_review" if needs_review else "send_reply"
print(f"Draft response: {response.content}")
return Command(
update={"draft_response": response.content}, # Store only the raw response
goto=goto
)
def human_review(state: EmailAgentState) -> Command[Literal["send_reply", END]]:
"""Pause for human review using interrupt and route based on decision"""
classification = state.get('classification', {})
# interrupt() must come first - any code before it will re-run on resume
human_decision = interrupt({
"email_id": state.get('email_id', ''),
"original_email": state.get('email_content', ''),
"draft_response": state.get('draft_response', ''),
"urgency": classification.get('urgency'),
"intent": classification.get('intent'),
"action": "Please review and approve/edit this response"
})
# Now process the human's decision
if human_decision.get("approved"):
return Command(
update={"draft_response": human_decision.get("edited_response", state.get('draft_response', ''))},
goto="send_reply"
)
else:
# Rejection means human will handle directly
return Command(update={}, goto=END)
def send_reply(state: EmailAgentState) -> Command[Literal["save_user_request"]]:
"""Send the email response"""
# Integrate with email service
print(f"Sending reply: {state['draft_response'][:100]}...")
return Command(
goto="save_user_request"
)
def save_user_request(state: EmailAgentState, store: PostgresStore) -> dict:
# 从state中获取email_id和classification数据
email_id = state.get('email_id')
classification_data = state.get('classification')
# 检查必要数据是否存在
if email_id and classification_data:
try:
# 使用store的put方法保存数据
# 将email_id作为键，classification_data作为值
store.put(("user_requests",), email_id, classification_data)
print(f"成功保存用户请求，邮件ID: {email_id}, 分类信息: {classification_data}")
except Exception as e:
# 捕获并打印可能发生的异常
print(f"保存用户请求时出错: {e}")
else:
# 如果缺少必要数据，打印警告信息
print(f"无法保存请求，缺少email_id或classification数据。State: {state}")
# 返回原始state，或者根据需要返回修改后的state
return {}
from langgraph.types import RetryPolicy
DB_URI = "postgresql://postgres:1314520@localhost:5432/Test?sslmode=disable"
with (
PostgresStore.from_conn_string(DB_URI) as store,
PostgresSaver.from_conn_string(DB_URI) as checkpointer,
):
store.setup()
checkpointer.setup()
# Create the graph
workflow = StateGraph(EmailAgentState)
# Add nodes with appropriate error handling
workflow.add_node("read_email", read_email)
workflow.add_node("classify_intent", classify_intent)
# Add retry policy for nodes that might have transient failures
workflow.add_node(
"search_documentation",
search_documentation,
retry_policy=RetryPolicy(max_attempts=3)
)
workflow.add_node("bug_tracking", bug_tracking)
workflow.add_node("draft_response", draft_response)
workflow.add_node("human_review", human_review)
workflow.add_node("send_reply", send_reply)
workflow.add_node("save_user_request", functools.partial(save_user_request, store=store))
# Add only the essential edges
workflow.add_edge(START, "read_email")
workflow.add_edge("read_email", "classify_intent")
workflow.add_edge("send_reply", "save_user_request")
workflow.add_edge("save_user_request", END)
app = workflow.compile(checkpointer=checkpointer, store=store)
# generate graph
generate_mermaid_image_advanced(app.get_graph().draw_mermaid())
# Test with an urgent billing issue
initial_state = {
"email_content": "i want to return the computer i bought last month, this is urgent",
"sender_email": "customer@example.com",
"email_id": "email_123",
"messages": []
}
# Run with a thread_id for persistence
config = {"configurable": {"thread_id": "customer_345"}}
result = app.invoke(initial_state, config)
# The graph will pause at human_review
print(f"human review interrupt:{result['__interrupt__']}")
# When ready, provide human input to resume
from langgraph.types import Command
human_response = Command(
resume={
"approved": True,
}
)
# Resume execution
final_result = app.invoke(human_response, config)
print(f"Email sent successfully!")
Persistence
关于持久化，我在langchain中已经介绍过了，我们常用的持久化有内存（InMemorySaver），数据库（PostgresSaver），Redis等，在本示例中，我们使用的是Postgres数据库，相关代码是：
with (
PostgresStore.from_conn_string(DB_URI) as store,
PostgresSaver.from_conn_string(DB_URI) as checkpointer,
):
store.setup()
checkpointer.setup()
#...
app = workflow.compile(checkpointer=checkpointer, store=store)
Checkpoint
The checkpointer uses thread_id as the primary key for storing and retrieving checkpoints. Checkpoints are persisted and can be used to restore the state of a thread at a later time.
在运行过我们的demo之后，我们会看到如下的数据库记录，记录了每个node在执行后产生的state数据变化，也就是snapshot，以及checkpoint相关的metadata。
Checkpoint使得人工介入（interrupt），或者失败重试成为可能，当我们需要从一个已知的checkpoint继续我们未完成的task时，我们可以用下面的方式：
# 从当前状态继续执行
config_with_checkpoint = {
"configurable": {
"thread_id": "test-1", # 任务唯一标识
"checkpoint_id": "1f0da204-f949-664a-bfff-3c4d4c4acfd1" # 需要断点继续的checkpointId
}
}
# 继续执行
result = await app.ainvoke(
None, # 不需要输入，从检查点继续
config=config_with_checkpoint
)
然而天下没有免费的午餐，高级功能的背后必有代价，如果我们查看第二步classify_intent的Checkpoint的内容，你会看到如下信息：
{
"v": 4,
"id": "1f0da205-29f1-672c-8002-d2a1d71bd4ae",
"ts": "2025-12-16T01:41:20.845188+00:00",
"versions_seen": {
"__input__": {},
"__start__": {
"__start__": "00000000000000000000000000000001.0.8181195432664937"
},
"read_email": {
"branch:to:read_email": "00000000000000000000000000000002.0.9889639973936442"
},
"classify_intent": {
"branch:to:classify_intent": "00000000000000000000000000000003.0.10211870765000752"
}
},
"channel_values": {
"email_id": "email_123",
"sender_email": "customer@example.com",
"email_content": "i want to return the computer i bought last month, this is urgent",
"branch:to:bug_tracking": null
},
"channel_versions": {
"email_id": "00000000000000000000000000000002.0.9889639973936442",
"messages": "00000000000000000000000000000003.0.10211870765000752",
"__start__": "00000000000000000000000000000002.0.9889639973936442",
"sender_email": "00000000000000000000000000000002.0.9889639973936442",
"email_content": "00000000000000000000000000000002.0.9889639973936442",
"classification": "00000000000000000000000000000004.0.4645694893583696",
"branch:to:read_email": "00000000000000000000000000000003.0.10211870765000752",
"branch:to:bug_tracking": "00000000000000000000000000000004.0.4645694893583696",
"branch:to:classify_intent": "00000000000000000000000000000004.0.4645694893583696"
},
"updated_channels": [
"branch:to:bug_tracking",
"classification"
]
}
LangGraph为了保证工作流的可恢复，可中断，使用了版本控制信息确保确定性重放和避免重复计算。实现这些“需求”给系统添加了很大的复杂度，我第一眼看到这些versions信息时，是一头雾水不知所云。另外，每一个node执行都会产生如此多的数据，规模大了，数据存储也将是个不小的问题。
Interrupt
Interrupt主要用于人工介入的场景，比如我们示例中的human_review节点，就是典型的人工介入场景，当interrupt( )被掉用的时候，整个workflow会暂定执行，直到接受到新的指令（Command），才能决定下一步的动作。其相关代码如下：
def human_review(state: EmailAgentState) -> Command[Literal["send_reply", END]]:
"""Pause for human review using interrupt and route based on decision"""
classification = state.get('classification', {})
# interrupt() invoked，workflow will pause here，waiting for human's decision Command
human_decision = interrupt({
"email_id": state.get('email_id', ''),
"original_email": state.get('email_content', ''),
"draft_response": state.get('draft_response', ''),
"urgency": classification.get('urgency'),
"intent": classification.get('intent'),
"action": "Please review and approve/edit this response"
})
# resume to process the human's decision
if human_decision.get("approved"):
return Command(
update={"draft_response": human_decision.get("edited_response", state.get('draft_response', ''))},
goto="send_reply"
)
else:
# Rejection means human will handle directly
return Command(update={}, goto=END)
在human_review这个节点中，整个工作流会在human_decision之后暂停，在执行下面代码之后。
config = {"configurable": {"thread_id": "customer_345"}}
human_response = Command(
resume={
"approved": True,
}
)
# Resume execution
final_result = app.invoke(human_response, config)
相当于给human_decision添加了新的变量"approved": True，这样workflow会按照命令走到send_reply，否则就直接结束了。
以上就是LangGraph的基本使用和核心概念的解释，对于这个简单的应用，还有一种更容易的实现方式，就是使用可视化的低代码平台。
可视化工作流
随着AI兴起，可用于 AI Agent 可视化工作流编排 的工具正在快速发展，这些工具帮助开发者以图形化方式设计、调试和部署基于大语言模型（LLM）的智能体（Agent），支持 多步推理、工具调用、记忆、条件分支、循环 等能力。可视化的低代码工作流编排工具、平台有很多，不管是通用的低代码的自动化工作流编排工具n8n（发音 “n-eight-n”），还是面向 AI 应用开发的低代码平台，专注于构建、部署和运营基于大语言模型（LLM）的 AI 应用的Dify。它们的形态都是大同小异，就是通过UI界面，拖拖拽拽，搞出一个Agent应用来。
对于我们客户邮件支持系统示例，如果要用n8n实现的话，我们可以在其UI界面上通过选择不同的节点，连线，形成一个如下的workflow，其实现的功能和我们通过一堆基于LangGraph代码实现的功能类似。
实际上，低代码也好，工作流也好，并不是什么新鲜玩意。“工作流” 本是上世纪 BPM（业务流程管理）的旧题。在AI兴起之前，为了提升“研发效率”，业界早就进行了无数次尝试，只是鲜有成功案例而已，其原因就在于这玩意也只能搞搞demo，一旦遇到复杂的生产场景，就会碰到Low Code Ceiling问题，一不小心，就会陷进万劫不复的深渊。关于这一点，想一下阿里的中台就明白了。
| 适用场景（推荐使用） | 不适用场景（慎用或需结合代码） |
|---|---|
| 快速构建 PoC / MVP | 高性能、低延迟的生产系统 |
| 规则明确的线性/分支流程 | 复杂动态逻辑（如强化学习式决策） |
| AI + SaaS 工具集成（邮件、CRM、数据库） | 需深度定制 LLM 推理逻辑 |
| 团队协作、非技术成员参与设计 | 需要严格测试覆盖和审计 |
| 中低频任务（如客服、内部工具） | 高并发、关键业务系统（如金融交易） |
太阳底下无新事
“太阳底下无新事”——无论是 LangGraph 的状态图、n8n 的节点连线，还是 Dify 的可视化编排，本质上都是对控制流与数据流这一古老计算范式的现代封装。它们以更友好的界面、更高的抽象层级，降低了构建 AI Agent 的初始门槛，让“人人皆可造智能体”成为可能。这无疑是进步，也是普惠。
然而，抽象总有代价。当工作流从演示走向真实业务，从日均百次调用迈向高并发、低延迟、强一致性的生产环境，低代码的“甜蜜外衣”之下，便显露出其结构性局限：难以调试的黑盒逻辑、脆弱的错误传播链、缺失的工程化能力（如灰度发布、性能压测、SLO 监控），以及最关键的——无法突破的“low-code ceiling”。你无法用拖拽解决 Token 爆炸、无法用图形节点实现高效的并发工具调用，更难在可视化界面中注入细粒度的安全审计或容灾回滚机制。
因此，作为公司的决策者，特别是架构师或工程师，请清醒地认识到：低代码是探索的起点，而非生产的终点。对于核心业务系统、关键客户交互、或任何对可靠性与可维护性有严苛要求的场景，真正的答案仍藏在代码之中——通过 LangGraph 这类可编程框架，结合严谨的软件工程实践，才能构建出既智能又健壮的下一代 Agent 系统。毕竟，再美的流程图，也替代不了一个经过压力测试、日志完备、可被团队集体维护的代码库。
最后
对于正在迷茫择业、想转行提升，或是刚入门的程序员、编程小白来说，有一个问题几乎人人都在问：未来10年，什么领域的职业发展潜力最大？
答案只有一个：人工智能（尤其是大模型方向）
当下，人工智能行业正处于爆发式增长期，其中大模型相关岗位更是供不应求，薪资待遇直接拉满——字节跳动作为AI领域的头部玩家，给硕士毕业的优质AI人才（含大模型相关方向）开出的月基础工资高达5万—6万元；即便是非“人才计划”的普通应聘者，月基础工资也能稳定在4万元左右。
再看阿里、腾讯两大互联网大厂，非“人才计划”的AI相关岗位应聘者，月基础工资也约有3万元，远超其他行业同资历岗位的薪资水平，对于程序员、小白来说，无疑是绝佳的转型和提升赛道。
对于想入局大模型、抢占未来10年行业红利的程序员和小白来说，现在正是最好的学习时机：行业缺口大、大厂需求旺、薪资天花板高，只要找准学习方向，稳步提升技能，就能轻松摆脱“低薪困境”，抓住AI时代的职业机遇。
如果你还不知道从何开始，我自己整理一套全网最全最细的大模型零基础教程，我也是一路自学走过来的，很清楚小白前期学习的痛楚，你要是没有方向还没有好的资源，根本学不到东西！
下面是我整理的大模型学习资源，希望能帮到你。
👇👇扫码免费领取全部内容👇👇
最后
1、大模型学习路线
2、从0到进阶大模型学习视频教程
从入门到进阶这里都有，跟着老师学习事半功倍。
3、 入门必看大模型学习书籍&文档.pdf（书面上的技术书籍确实太多了，这些是我精选出来的，还有很多不在图里）
4、 AI大模型最新行业报告
2026最新行业报告，针对不同行业的现状、趋势、问题、机会等进行系统地调研和评估，以了解哪些行业更适合引入大模型的技术和应用，以及在哪些方面可以发挥大模型的优势。
5、面试试题/经验
【大厂 AI 岗位面经分享（107 道）】
【AI 大模型面试真题（102 道）】
【LLMs 面试真题（97 道）】
6、大模型项目实战&配套源码
适用人群
四阶段学习规划（共90天，可落地执行）
第一阶段（10天）：初阶应用
该阶段让大家对大模型 AI有一个最前沿的认识，对大模型 AI 的理解超过 95% 的人，可以在相关讨论时发表高级、不跟风、又接地气的见解，别人只会和 AI 聊天，而你能调教 AI，并能用代码将大模型和业务衔接。
- 大模型 AI 能干什么？
- 大模型是怎样获得「智能」的？
- 用好 AI 的核心心法
- 大模型应用业务架构
- 大模型应用技术架构
- 代码示例：向 GPT-3.5 灌入新知识
- 提示工程的意义和核心思想
- Prompt 典型构成
- 指令调优方法论
- 思维链和思维树
- Prompt 攻击和防范
- …
第二阶段（30天）：高阶应用
该阶段我们正式进入大模型 AI 进阶实战学习，学会构造私有知识库，扩展 AI 的能力。快速开发一个完整的基于 agent 对话机器人。掌握功能最强的大模型开发框架，抓住最新的技术进展，适合 Python 和 JavaScript 程序员。
- 为什么要做 RAG
- 搭建一个简单的 ChatPDF
- 检索的基础概念
- 什么是向量表示（Embeddings）
- 向量数据库与向量检索
- 基于向量检索的 RAG
- 搭建 RAG 系统的扩展知识
- 混合检索与 RAG-Fusion 简介
- 向量模型本地部署
- …
第三阶段（30天）：模型训练
恭喜你，如果学到这里，你基本可以找到一份大模型 AI相关的工作，自己也能训练 GPT 了！通过微调，训练自己的垂直大模型，能独立训练开源多模态大模型，掌握更多技术方案。
到此为止，大概2个月的时间。你已经成为了一名“AI小子”。那么你还想往下探索吗？
- 为什么要做 RAG
- 什么是模型
- 什么是模型训练
- 求解器 & 损失函数简介
- 小实验2：手写一个简单的神经网络并训练它
- 什么是训练/预训练/微调/轻量化微调
- Transformer结构简介
- 轻量化微调
- 实验数据集的构建
- …
第四阶段（20天）：商业闭环
对全球大模型从性能、吞吐量、成本等方面有一定的认知，可以在云端和本地等多种环境下部署大模型，找到适合自己的项目/创业方向，做一名被 AI 武装的产品经理。
-
硬件选型
-
带你了解全球大模型
-
使用国产大模型服务
-
搭建 OpenAI 代理
-
热身：基于阿里云 PAI 部署 Stable Diffusion
-
在本地计算机运行大模型
-
大模型的私有化部署
-
基于 vLLM 部署大模型
-
案例：如何优雅地在阿里云私有部署开源大模型
-
部署一套开源 LLM 项目
-
内容安全
-
互联网信息服务算法备案
-
…
👇👇扫码免费领取全部内容👇👇
3、这些资料真的有用吗？
这份资料由我和鲁为民博士(北京清华大学学士和美国加州理工学院博士)共同整理，现任上海殷泊信息科技CEO，其创立的MoPaaS云平台获Forrester全球’强劲表现者’认证，服务航天科工、国家电网等1000+企业，以第一作者在IEEE Transactions发表论文50+篇，获NASA JPL火星探测系统强化学习专利等35项中美专利。本套AI大模型课程由清华大学-加州理工双料博士、吴文俊人工智能奖得主鲁为民教授领衔研发。
资料内容涵盖了从入门到进阶的各类视频教程和实战项目，无论你是小白还是有些技术基础的技术人员，这份资料都绝对能帮助你提升薪资待遇，转行大模型岗位。
这份完整版的大模型 AI 学习资料已经上传CSDN，朋友们如果需要可以微信扫描下方CSDN官方认证二维码免费领取【保证100%免费】
有“AI”的1024 = 2048，欢迎大家加入2048 AI社区
更多推荐
4
0- 0
已为社区贡献465条内容
所有评论(0)