【必学收藏】LangGraph vs 低代码平台：构建生产级AI Agent的终极选择指南

本文系统介绍了LangGraph框架与可视化低代码平台(n8n/Dify)在构建AI Agent时的对比分析。LangGraph作为底层编排框架，提供精细控制能力，适合构建复杂、生产级的智能系统，具备状态管理、检查点和人工中断等高级功能。而低代码平台通过可视化界面降低开发门槛，适合快速原型和简单场景。文章强调低代码是探索起点而非生产终点，核心业务系统仍需代码级实现以确保可靠性和可维护性。

LangGraph

LangGraph is alow-level orchestration frameworkand 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 usesthread_idas 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问题，一不小心，就会陷进万劫不复的深渊。关于这一点，想一下阿里的中台就明白了。

太阳底下无新事

“太阳底下无新事”——无论是 LangGraph 的状态图、n8n 的节点连线，还是 Dify 的可视化编排，本质上都是对控制流与数据流这一古老计算范式的现代封装。它们以更友好的界面、更高的抽象层级，降低了构建 AI Agent 的初始门槛，让“人人皆可造智能体”成为可能。这无疑是进步，也是普惠。

然而，抽象总有代价。当工作流从演示走向真实业务，从日均百次调用迈向高并发、低延迟、强一致性的生产环境，低代码的“甜蜜外衣”之下，便显露出其结构性局限：难以调试的黑盒逻辑、脆弱的错误传播链、缺失的工程化能力（如灰度发布、性能压测、SLO 监控），以及最关键的——无法突破的“low-code ceiling”。你无法用拖拽解决 Token 爆炸、无法用图形节点实现高效的并发工具调用，更难在可视化界面中注入细粒度的安全审计或容灾回滚机制。

因此，作为公司的决策者，特别是架构师或工程师，请清醒地认识到：低代码是探索的起点，而非生产的终点。对于核心业务系统、关键客户交互、或任何对可靠性与可维护性有严苛要求的场景，真正的答案仍藏在代码之中——通过 LangGraph 这类可编程框架，结合严谨的软件工程实践，才能构建出既智能又健壮的下一代 Agent 系统。毕竟，再美的流程图，也替代不了一个经过压力测试、日志完备、可被团队集体维护的代码库。

如何学习大模型 AI ？

由于新岗位的生产效率，要优于被取代岗位的生产效率，所以实际上整个社会的生产效率是提升的。

但是具体到个人，只能说是：

“最先掌握AI的人，将会比较晚掌握AI的人有竞争优势”。

这句话，放在计算机、互联网、移动互联网的开局时期，都是一样的道理。

我在一线互联网企业工作十余年里，指导过不少同行后辈。帮助很多人得到了学习和成长。

我意识到有很多经验和知识值得分享给大家，也可以通过我们的能力和经验解答大家在人工智能学习中的很多困惑，所以在工作繁忙的情况下还是坚持各种整理和分享。但苦于知识传播途径有限，很多互联网行业朋友无法获得正确的资料得到学习提升，故此将并将重要的AI大模型资料包括AI大模型入门学习思维导图、精品AI大模型学习书籍手册、视频教程、实战学习等录播视频免费分享出来。

这份完整版的大模型 AI 学习资料已经上传CSDN，朋友们如果需要可以微信扫描下方CSDN官方认证二维码免费领取【保证100%免费】

为什么要学习大模型？

我国在A大模型领域面临人才短缺,数量与质量均落后于发达国家。2023年，人才缺口已超百万，凸显培养不足。随着AI技术飞速发展，预计到2025年,这一缺口将急剧扩大至400万,严重制约我国AI产业的创新步伐。加强人才培养,优化教育体系,国际合作并进是破解困局、推动AI发展的关键。

大模型入门到实战全套学习大礼包

1、大模型系统化学习路线

作为学习AI大模型技术的新手，方向至关重要。 正确的学习路线可以为你节省时间，少走弯路；方向不对，努力白费。这里我给大家准备了一份最科学最系统的学习成长路线图和学习规划，带你从零基础入门到精通！

2、大模型学习书籍&文档

学习AI大模型离不开书籍文档，我精选了一系列大模型技术的书籍和学习文档（电子版），它们由领域内的顶尖专家撰写，内容全面、深入、详尽，为你学习大模型提供坚实的理论基础。

3、AI大模型最新行业报告

2025最新行业报告，针对不同行业的现状、趋势、问题、机会等进行系统地调研和评估，以了解哪些行业更适合引入大模型的技术和应用，以及在哪些方面可以发挥大模型的优势。

4、大模型项目实战&配套源码

学以致用，在项目实战中检验和巩固你所学到的知识，同时为你找工作就业和职业发展打下坚实的基础。

5、大模型大厂面试真题

面试不仅是技术的较量，更需要充分的准备。在你已经掌握了大模型技术之后，就需要开始准备面试，我精心整理了一份大模型面试题库，涵盖当前面试中可能遇到的各种技术问题，让你在面试中游刃有余。

适用人群

第一阶段（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 项目
• 内容安全
• 互联网信息服务算法备案
• …

学习是一个过程，只要学习就会有挑战。天道酬勤，你越努力，就会成为越优秀的自己。

如果你能在15天内完成所有的任务，那你堪称天才。然而，如果你能完成 60-70% 的内容，你就已经开始具备成为一名大模型 AI 的正确特征了。

这份完整版的大模型 AI 学习资料已经上传CSDN，朋友们如果需要可以微信扫描下方CSDN官方认证二维码免费领取【保证100%免费】