LangGraph实践-构建AI工作流:创建一本大模型应用开发书籍
如果你想使用dify，coze以及n8n构建你的工作流，那基本上非常轻松，拖拽组件即可完成，但是因为基本上都是低代码平台，因此你可以操作的空间非常小，当然据我了解n8n是可以支持js或者python添加节点的，这为工作流的灵活性开了窗口，使可以会写代码的人有了自由发挥的空间，但是如果全流程想要代码实现，那就得用LangGraph这样的框架去实现了。
LangGraph 是由 LangChain 团队开发的一个开源框架，专为构建有状态、多智能体（Multi-Agent）的 AI 应用而设计。它通过图结构（Graph）来定义和协调复杂的工作流，特别适合需要循环、分支、持久化状态管理和人工干预的场景。
LangGraph 的核心特点
1、图结构（Graph-Based）
- • 采用有向图（Directed Graph）模型，由节点（Nodes）和边（Edges）组成：
- • 节点：代表任务单元（如调用 LLM、执行工具函数、数据库查询等）。• 边：定义节点间的流转逻辑，支持条件分支和循环。
2、状态管理（State Management）
- • 提供全局共享状态（State），支持跨节点数据传递和持久化存储。• 支持断点续跑（Checkpointing），即使任务中断也能恢复执行。
3、循环与分支（Loops & Conditional Flows）
- • 允许 AI 代理根据中间结果动态调整执行路径（如反复优化答案直至满意）。• 适用于RAG（检索增强生成）、代码优化、多轮决策等场景。
4、人类在环（Human-in-the-Loop）
- • 支持人工干预，如审核 AI 决策、修改中间结果，适用于医疗、法律等专业领域。
5、多智能体协作（Multi-Agent Systems）
- • 可构建分工明确的 AI 代理团队（如“规划代理”制定策略，“执行代理”调用工具。
6、与 LangChain 无缝集成
- • 可直接复用 LangChain 的 Chains、Tools 和 Memory 组件，同时提供更强大的流程控。
LangGraph 的典型应用场景
- • 复杂问答系统（如带多轮优化的客服机器人）。• 代码生成与测试（如自动生成代码并迭代优化）。• 金融风控 & 数据分析（如动态调整决策流程）。• 多智能体协作（如多个 AI 代理协同完成任务）。• 医疗诊断辅助（结合专家审核的 AI 决策系统。
LangGraph vs. Dify/Coze/n8n
| 维度 | LangGraph | Dify | Coze | n8n |
|---|---|---|---|---|
| 架构 | 基于图结构（节点+边），支持循环/分支，专为多Agent协作设计 | 低代码AI应用开发平台，支持RAG & 复杂工作流 | 零代码对话式AI开发，拖拽式流程编排 | 节点式工作流自动化，支持跨系统集成 |
| 状态管理 | 全局State + 持久化存储，支持断点续跑 | 会话级上下文管理 | 对话状态跟踪 | 无原生状态管理，需外部存储 |
| 使用场景 | 复杂决策流（多轮优化、Agent协作） | 企业级AI应用（知识库/RAG） | 轻量级对话机器人（客服/社交） | 跨系统自动化（数据同步/API调用） |
| 学习成本 | 高（需Python编码，理解图结构） | 中（需技术背景，熟悉LLM） | 低（零代码，适合非技术用户） | 中（需API/逻辑理解） |
| 开源/商业 | 开源（MIT协议） | 开源（部分企业功能付费） | 部分开源（Apache 2.0） | 完全开源（可持续使用协议） |
选型建议
- 1. LangGraph：
- • 适用：需要复杂AI工作流（如多Agent协作、动态路由、循环优化）的开发团队。• 优势：灵活性强，支持持久化状态，适合金融、医疗等专业领域。
- • 适用：企业级AI应用（如知识库问答、多模型集成），需私有化部署。• 优势：低代码开发，支持RAG，适合技术团队快速构建AI应用。
- • 适用：快速搭建对话机器人（如抖音/飞书集成），适合中小团队或个人开发者。• 优势：零代码体验，内置字节生态插件，部署简单。
- • 适用：非AI自动化（如数据同步、API调用），技术团队自托管需求。• 优势：开源免费，扩展性强，支持1000+第三方服务集成。
组合使用建议
- • AI + 自动化：Dify（AI核心） + n8n（流程自动化）。• 复杂Agent系统：LangGraph（逻辑编排） + Coze（前端交互）。
如需更详细的功能对比或部署指南，可参考各平台的官方文档或社区讨论。
LangGraph实战
用LangGraph开发一个工作流，用于创建一本大模型应用的书籍。
要开始使用 LangGraph，首先需要安装必要的包：pip install langchain langgraph langchain-community
如果你想使用可视化和监控功能，还需要安装 LangSmith：pip install langsmith
然后就是登录https://smith.langchain.com/，创建一个项目，然后generate key，点击复制这个key，填到代码中对应这个key下os.environ["LANGCHAIN_API_KEY"] = "your_langsmith_api_key" # 替换为你的 API Key
最后就是下载安装ollama，然后通过ollama安装llama3.2模型：
➜ ~ ollama run llama3.2pulling manifestpulling dde5aa3fc5ff: 100% ▕████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████▏ 2.0 GBpulling 966de95ca8a6: 100% ▕████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████▏ 1.4 KBpulling fcc5a6bec9da: 100% ▕████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████▏ 7.7 KBpulling a70ff7e570d9: 100% ▕████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████▏ 6.0 KBpulling 56bb8bd477a5: 100% ▕████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████▏ 96 Bpulling 34bb5ab01051: 100% ▕████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████▏ 561 Bverifying sha256 digestwriting manifestsuccess>>>安装完成之后，我们可以测试一下，那我的测试语句就是：
>>> who are youI'm an artificial intelligence model known as Llama. Llama stands for "Large Language Model Meta AI.">>>证明模型是安装成功的，接下来开始写代码。
现在我贴出代码：
import osimport timeimport requestsfrom typing import Dict, List, Any, TypedDictfrom langgraph.graph import StateGraph# Ollama 配置OLLAMA_API_BASE = "http://localhost:11434/api" # ollama提供访问模型的地址DEFAULT_MODEL = "llama3.2" # 这就是我们刚才安装的llama模型，注意必须是3.2# LangSmith 配置os.environ["LANGCHAIN_TRACING_V2"] = "true"os.environ["LANGCHAIN_API_KEY"] = "your_langsmith_api_key" # 替换为你的 API Keyos.environ["LANGCHAIN_PROJECT"] = "课程创建-LangGraph"# Ollama API 调用函数def generate_with_ollama(prompt, model=DEFAULT_MODEL, temperature=0.5, max_tokens=2000): """使用 Ollama API 生成内容""" try: response = requests.post( f"{OLLAMA_API_BASE}/generate", json={ "model": model, "prompt": prompt, "temperature": temperature, "stream": False } ) if response.status_code == 200: result = response.json() return result.get("response", "") else: print(f"Ollama API 返回错误状态码: {response.status_code}") return f"API调用失败，状态码: {response.status_code}" except Exception as e: print(f"调用 Ollama API 出错: {e}") return f"内容生成失败，请检查 Ollama 服务。错误: {e}"# 课程结构定义course_structure = { "课程名称": "大模型应用开发从入门到放弃", "章节": [ { "章节名称": "大模型应用开发具备技术栈", "简介": "介绍大模型应用开发必须掌握和了解的基础知识和工具", "学习目标": ["理解大模型应用开发的基本概念", "掌握大模型应用的基础开发流程"] }, { "章节名称": "大模型应用实战", "简介": "用大模型+LangGraph框架开发一个hello world示例", "学习目标": ["夯实大模型的基础理论", "掌握大模型应用开发的基础框架"] } ]}# 提示词模板prompt_templates = { "概念理解": """ 你是一位大模型应用开发专家，正在为《{course_name}》课程的《{chapter_name}》章节创建概念理解学习材料。 章节介绍: {chapter_intro} 学习目标: {chapter_goals} 请创建一份全面的概念理解材料，包含核心概念定义、背景知识、相关概念联系等。 输出格式使用 Markdown，确保标题层级正确，内容应该全部使用中文。 """, "实践应用": """ 你是一位大模型应用开发专家，正在为《{course_name}》课程的《{chapter_name}》章节创建实践应用学习材料。 章节介绍: {chapter_intro} 学习目标: {chapter_goals} 请创建一份全面的实践应用材料，对大模型应用开发提供一个完整的实践案例。 输出格式使用 Markdown，确保标题层级正确，内容应该全部使用中文。 """, "知识评估": """ 你是一位大模型应用开发专家，正在为《{course_name}》课程的《{chapter_name}》章节创建知识评估学习材料。 章节介绍: {chapter_intro} 学习目标: {chapter_goals} 请创建一份全面的知识评估材料，对大模型应用查出的结果进行评估。 输出格式使用 Markdown，确保标题层级正确，内容应该全部使用中文。 """}# 1. 章节生成节点def chapter_generator(state: Dict[str, Any]) -> Dict[str, Any]: """生成课程章节""" print("章节生成节点开始执行...") # 简化版：直接使用预定义的章节 chapters = course_structure["章节"] # 返回更新后的完整状态 updated_state = state.copy() updated_state["章节列表"] = chapters print(f"章节生成完成，返回状态键: {list(updated_state.keys())}") return updated_state# 2. 学习对象生成节点def learning_object_generator(state: Dict[str, Any]) -> Dict[str, Any]: """为每个章节生成学习对象""" print("学习对象生成节点开始执行...") print(f"接收到的状态键: {list(state.keys())}") chapters = state["章节列表"] learning_objects = [] for chapter in chapters: chapter_name = chapter["章节名称"] # 创建基础学习对象 chapter_objects = { "章节名称": chapter_name, "学习对象": [ {"类型": "概念理解", "内容": f"本节介绍{chapter_name}的核心概念，帮助学习者建立基础认知。"}, {"类型": "实践应用", "内容": f"通过实际案例演示{chapter_name}的应用方法，提升实战能力。"}, {"类型": "知识评估", "内容": f"包含与{chapter_name}相关的自测题，帮助学习者检验学习效果。"} ] } learning_objects.append(chapter_objects) # 返回更新后的完整状态 updated_state = state.copy() updated_state["章节学习对象"] = learning_objects print(f"学习对象生成完成，返回状态键: {list(updated_state.keys())}") return updated_state# 3. 内容增强节点def content_enhancer(state: Dict[str, Any]) -> Dict[str, Any]: """使用 Ollama 生成详细内容""" print("开始内容增强...") print(f"当前状态键: {list(state.keys())}") # 检查章节列表是否存在 if "章节列表" not in state: print("错误: 状态中没有找到'章节列表'") return state chapter_list = state["章节列表"] learning_objects_list = state.get("章节学习对象", []) # 创建映射以便后续查找 learning_objects_map = {} for lo in learning_objects_list: chapter_name = lo.get("章节名称", "") if chapter_name: learning_objects_map[chapter_name] = lo.get("学习对象", []) # 存储增强后的章节内容 enhanced_chapters = [] # 处理每个章节 for chapter in chapter_list: try: # 提取章节信息 chapter_name = chapter["章节名称"] chapter_intro = chapter.get("简介", "无介绍") # 处理学习目标 chapter_goals = chapter.get("学习目标", []) if isinstance(chapter_goals, list): chapter_goals = "\n".join([f"- {goal}" for goal in chapter_goals]) # 创建增强章节对象 enhanced_chapter = { "章节名称": chapter_name, "章节介绍": chapter_intro, "学习目标": chapter_goals, "学习对象": [] } # 获取学习对象类型 learning_objects = [] if chapter_name in learning_objects_map: chapter_objects = learning_objects_map[chapter_name] learning_objects = [obj["类型"] for obj in chapter_objects] # 如果没有定义学习对象，使用默认列表 if not learning_objects: learning_objects = ["概念理解", "实践应用", "知识评估"] # 为每个学习对象生成内容 for obj_type in learning_objects: # 创建提示词 prompt = prompt_templates.get(obj_type, "").format( course_name=course_structure["课程名称"], chapter_name=chapter_name, chapter_intro=chapter_intro, chapter_goals=chapter_goals ) # 使用 Ollama 生成内容 enhanced_content = generate_with_ollama(prompt) # 添加到章节对象 enhanced_chapter["学习对象"].append({ "类型": obj_type, "内容": enhanced_content }) # 添加到增强章节列表 enhanced_chapters.append(enhanced_chapter) except Exception as e: print(f"处理章节时发生错误: {e}") # 更新状态 updated_state = state.copy() updated_state["章节增强内容"] = enhanced_chapters return updated_state# 4. Markdown 写入节点def markdown_writer(state: Dict[str, Any]) -> Dict[str, Any]: """保存章节与学习对象到本地 Markdown 文件""" course_name = course_structure["课程名称"] enhanced_chapters = state.get("章节增强内容", []) # 创建输出目录 output_dir = f"course_langgraph_chapter_1" os.makedirs(output_dir, exist_ok=True) # 创建索引文件 index_path = os.path.join(output_dir, "课程索引.md") with open(index_path, "w", encoding="utf-8") as f: f.write(f"# {course_name}\n\n") f.write("## 课程章节\n\n") for chapter in enhanced_chapters: chapter_name = chapter["章节名称"] chapter_intro = chapter["章节介绍"] chapter_goals = chapter["学习目标"] # 写入章节信息 f.write(f"### {chapter_name}\n\n") f.write(f"{chapter_intro}\n\n") f.write("**学习目标**:\n") if isinstance(chapter_goals, list): for goal in chapter_goals: f.write(f"- {goal}\n") else: f.write(f"{chapter_goals}\n") # 创建章节目录 chapter_dir = os.path.join(output_dir, chapter_name) os.makedirs(chapter_dir, exist_ok=True) # 添加学习对象链接 f.write("\n**学习内容**:\n") # 写入每个学习对象 for obj in chapter["学习对象"]: obj_type = obj["类型"] obj_content = obj["内容"] # 写入文件链接 obj_filename = f"{obj_type}.md" obj_path = os.path.join(chapter_dir, obj_filename) f.write(f"- [{obj_type}](./{chapter_name}/{obj_filename})\n") # 创建学习对象文件 with open(obj_path, "w", encoding="utf-8") as obj_file: obj_file.write(f"# {chapter_name} - {obj_type}\n\n") obj_file.write(obj_content) f.write("\n") return state# 创建工作流图builder = StateGraph(Dict[str, Any])# 添加节点builder.add_node("章节生成", chapter_generator)builder.add_node("学习对象生成", learning_object_generator)builder.add_node("内容增强", content_enhancer)builder.add_node("Markdown写入", markdown_writer)# 设置起点builder.set_entry_point("章节生成")# 添加边（连接节点）builder.add_edge("章节生成", "学习对象生成")builder.add_edge("学习对象生成", "内容增强")builder.add_edge("内容增强", "Markdown写入")# 编译图workflow = builder.compile(debug=True)# 执行工作流if __name__ == "__main__": # 初始状态为空 state = {} # 添加运行标识 run_id = f"run_{int(time.time())}" print(f"开始执行课程生成流程，运行ID: {run_id}") print(f"可在 LangSmith 查看详细执行记录: https://smith.langchain.com/projects/课程创建-LangGraph") # 执行工作流 result = workflow.invoke(state) print(f"执行完成! 课程内容已生成。")结果展示
生成的章节目录结构图：
我们点击书籍索引文件：
随便点击一个章节下的小结，我们点击大模型应用开发具备技术栈章下面的概念理解小节：
然后我们再看下实践应用小节：
非常完美的实现了一个简化版本的AI工作流。
小结
LangGraph 是构建复杂、有状态AI工作流的强大工具，尤其适合需要循环、多智能体协作和人工干预的场景。相比 Dify/Coze/n8n，它更适合开发者需要精细化控制的AI应用。
希望接下来大家也尝试写出自己的AI工作流，助力工作提效或者个人成长！
历史大模型系列文章：
- DeepSeek大模型之本地部署体验提示工程(prompt engineering)大揭秘OpenAI：GPT提示词(prompt)六大策略之解读GPT prompt(提示词)实战，用李佳琪式的流行梗回复，有点意思提示词(prompt)那些事彻底理解推理模型和通用模型基于LangChain进行大模型应用开发-第一节基于LangChain进行大模型应用开发-第二节(检索器)基于LangChain进行大模型应用开发-第三节(对话检索器(聊天机器人))基于LangChain进行大模型应用开发-第四节(agent代理)基于LangChain进行大模型应用开发-第五节(LangServe服务开发)LangStudio构建nl2sql应用流，只要输入文字就可以转为sql执行并输出结果解密manus的神秘面纱-搜索JackMa演示多智能体(Agent)的强大无比camel-ai推出的OWL用于处理现实世界的自动化任务大模型怎么知道铁是导电的？LangStudio构建文件应用流，只要输入地址就可以转为你自己的知识库AI大模型学习框架基于百炼构建天气预报应用，非常简单基于MCP构建天气预报智能体《零代码+AI赋能！基于阿里云百炼MCP快速打造智能爬虫应用》AI时代的高效信息管家：基于MCP-Agent与通义千问的网页总结智能体实战manus终于开放啦在Mac上10分钟玩转LoRA微调模型揭秘Google A2A协议：赋能智能体协作的未来学习【北京大学】DeepSeek内部研讨系列: AI Agent与Agentic AI的原理和应用并做分享提示工程(Prompt Engineering)还没学明白，上下文工程(Context Engineering)又来了深度剖析Manus：如何打造低幻觉、高效率、安全可靠的Agentic AI系统大模型时代的三巨头—Grok、ChatGPT与Gemini深度解析深度解析AI Agent：从概念、原理到构建模式，一文彻底搞懂！打造一个微claude code来实现石头剪刀布游戏豆包新模型+PromptPilot体验测评Google Agent白皮书：深度解析生成式AI Agent
