2026/4/17 16:52:23
网站建设
项目流程
网站视频放优酷里面怎么做,百度推广点击收费标准,开平市建设工程站网站,asp网站优化一、LangGraph 核心概念拆解#xff08;理解工作流的 “骨架”#xff09;
视频首先系统讲解 LangGraph 的核心组件与设计思想#xff0c;避免开发者因概念模糊导致后续开发踩坑#xff0c;核心围绕 “状态#xff08;State#xff09;、节点#xff08;Node#xff0…一、LangGraph 核心概念拆解理解工作流的 “骨架”视频首先系统讲解 LangGraph 的核心组件与设计思想避免开发者因概念模糊导致后续开发踩坑核心围绕 “状态State、节点Node、边Edge、图Graph” 四大核心要素展开一核心组件定义与作用组件定义与核心作用关键特性状态State工作流数据流转的 “载体”存储工作流执行过程中的所有数据输入、中间结果、配置参数1. 基于类 / 数据结构定义支持任意字段扩展2. 节点间通过修改 State 传递数据3. 支持状态持久化与回溯节点Node工作流的 “执行单元”封装具体业务逻辑如数据处理、模型调用、工具调用1. 输入输出均为 State 实例2. 支持同步 / 异步函数3. 可复用、可组合支持自定义节点与内置节点边Edge节点间的 “连接关系”定义数据流转路径如 “节点 A 执行完成→节点 B”1. 分为 “普通边”固定流转与 “条件边”动态流转2. 支持单节点→多节点、多节点→单节点连接图Graph工作流的 “容器”整合 State、Node、Edge形成完整的执行逻辑1. 需指定 “入口节点”Entry Point与 “结束节点”Finish Point2. 编译后生成可执行的 App 实例二核心设计思想“状态驱动的有向图”状态驱动所有节点的执行逻辑围绕 State 展开节点通过读取 State 中的数据进行计算执行完成后修改 State 并传递给下一个节点确保数据流转的一致性有向图结构工作流的执行路径由 Edge 明确定义支持线性、分支、循环等复杂结构可视化程度高便于理解与调试可扩展性组件化设计允许灵活新增 Node、Edge或扩展 State 字段适配不同业务场景。二、线性工作流开发实操入门必学视频以 “文本处理工作流” 为例演示线性工作流无分支、无循环节点按固定顺序执行的完整开发流程步骤清晰可复刻一开发步骤从定义到执行步骤 1定义 State数据载体基于 LangGraph 的State类定义工作流所需的所有数据字段示例python运行from langgraph.graph import State from typing import Optional class TextProcessState(State): # 输入数据 raw_text: str # 原始文本必填 # 中间结果 cleaned_text: Optional[str] None # 清洗后的文本可选默认None summarized_text: Optional[str] None # 摘要文本可选默认None # 配置参数 summary_length: int 50 # 摘要长度默认50字关键说明字段需明确类型注解可选字段设置默认值确保数据类型一致性。步骤 2编写 Node 函数执行单元每个 Node 函数接收State实例作为输入修改后返回新的State示例实现两个节点文本清洗、文本摘要python运行# 节点1文本清洗去除空格、特殊字符 def clean_text_node(state: TextProcessState) - TextProcessState: # 读取State中的原始文本 raw_text state.raw_text # 执行清洗逻辑 cleaned raw_text.strip().replace(\n, ).replace( , ) # 修改State并返回 state.cleaned_text cleaned return state # 节点2文本摘要基于简单规则生成摘要 def summarize_text_node(state: TextProcessState) - TextProcessState: # 读取State中的清洗后文本与配置参数 cleaned_text state.cleaned_text summary_length state.summary_length # 执行摘要逻辑实际场景可替换为LLM调用 if len(cleaned_text) summary_length: summary cleaned_text else: summary cleaned_text[:summary_length] ... # 修改State并返回 state.summarized_text summary return state关键说明Node 函数必须返回State实例不可直接返回原始数据逻辑需围绕 State 中的字段展开。步骤 3构建 Graph整合节点与边创建 Graph 实例添加节点并定义流转顺序线性入口→清洗节点→摘要节点→结束python运行from langgraph import Graph # 1. 创建Graph实例指定State类型 graph Graph(TextProcessState) # 2. 添加节点参数节点ID节点函数 graph.add_node(clean_text, clean_text_node) graph.add_node(summarize_text, summarize_text_node) # 3. 定义边固定流转顺序 graph.add_edge(clean_text, summarize_text) # 清洗节点→摘要节点 # 4. 指定入口与结束节点 graph.set_entry_point(clean_text) # 入口清洗节点 graph.set_finish_point(summarize_text) # 结束摘要节点步骤 4编译与执行工作流将 Graph 编译为可执行的 App 实例传入初始 State 执行获取结果python运行# 1. 编译Graph app graph.compile() # 2. 定义初始State仅传入必填字段 initial_state TextProcessState( raw_text 这是一段需要处理的原始文本包含多余的空格和换行符。\n\n 下面是第二段内容我们需要对其进行清洗和摘要生成。 , summary_length80 # 自定义摘要长度覆盖默认值 ) # 3. 执行工作流invoke方法接收初始State返回最终State final_state app.invoke(initial_state) # 4. 查看结果 print(清洗后文本, final_state.cleaned_text) print(生成摘要, final_state.summarized_text)步骤 5执行结果验证预期输出plaintext清洗后文本 这是一段需要处理的原始文本包含多余的空格和换行符。 下面是第二段内容我们需要对其进行清洗和摘要生成。 生成摘要 这是一段需要处理的原始文本包含多余的空格和换行符。 下面是第二段内容我们需要对其进行清洗和摘要生成。...二关键注意事项节点函数必须接收并返回State实例不可遗漏边的连接必须存在对应的节点避免节点 ID 拼写错误初始 State 需包含所有必填字段可选字段可按需覆盖默认值。三、分支工作流开发实操进阶入门视频在 linear 工作流基础上扩展至分支工作流根据 State 中的数据动态选择执行路径以 “数据类型判断工作流” 为例演示条件边的使用一核心新增组件条件边ConditionalEdge作用根据 State 中的数据动态选择下一个执行节点实现 “一进多出” 的分支逻辑核心依赖需定义 “分支判断函数”输入 State返回目标节点 ID。二开发步骤分支工作流实现步骤 1扩展 State增加分支判断字段python运行class DataProcessState(State): data: str # 输入数据 data_type: Optional[str] None # 数据类型text/image/other result: Optional[str] None # 处理结果步骤 2编写分支判断函数python运行def route_by_data_type(state: DataProcessState) - str: 根据数据类型选择下一个节点 data state.data # 简单判断数据类型实际场景可替换为更复杂的识别逻辑 if data.startswith(text:): state.data_type text return process_text # 文本处理节点ID elif data.startswith(image:): state.data_type image return process_image # 图像处理节点ID else: state.data_type other return process_other # 其他类型处理节点ID步骤 3编写分支节点函数python运行# 文本处理节点 def process_text_node(state: DataProcessState) - DataProcessState: state.result f文本处理完成{state.data[5:]} # 截取text:后的内容 return state # 图像处理节点 def process_image_node(state: DataProcessState) - DataProcessState: state.result f图像处理完成{state.data[6:]} # 截取image:后的内容 return state # 其他类型处理节点 def process_other_node(state: DataProcessState) - DataProcessState: state.result f其他类型数据处理完成{state.data} return state步骤 4构建分支 Graphpython运行graph Graph(DataProcessState) # 添加所有节点 graph.add_node(route, route_by_data_type) # 分支判断节点可选也可直接用条件边 graph.add_node(process_text, process_text_node) graph.add_node(process_image, process_image_node) graph.add_node(process_other, process_other_node) # 定义条件边入口→分支判断→目标节点 graph.add_conditional_edges( sourceroute, # 源节点ID分支判断节点 conditionroute_by_data_type, # 分支判断函数 # 映射判断函数返回值→目标节点ID mapping{ process_text: process_text, process_image: process_image, process_other: process_other } ) # 指定入口节点分支判断节点 graph.set_entry_point(route) # 所有分支共享同一个结束节点可选也可分别设置 graph.set_finish_point(process_text) graph.set_finish_point(process_image) graph.set_finish_point(process_other)步骤 5执行与验证python运行app graph.compile() # 测试文本类型数据 text_state DataProcessState(datatext:这是一段文本数据) text_result app.invoke(text_state) print(text_result.result) # 预期输出文本处理完成这是一段文本数据 # 测试图像类型数据 image_state DataProcessState(dataimage:这是一张图片的路径) image_result app.invoke(image_state) print(image_result.result) # 预期输出图像处理完成这是一张图片的路径四、调试与优化技巧提升开发效率视频针对 LangGraph 开发中的调试痛点提供了 3 个核心技巧一日志输出与状态查看启用详细日志通过logging模块输出节点执行顺序、State 变化便于定位问题python运行import logging logging.basicConfig(levellogging.INFO) # 输出INFO级别日志执行后会打印节点执行日志如 “Entering node clean_text”“Exiting node clean_text”。中间状态查看在节点函数中添加print语句实时查看 State 的变化python运行def clean_text_node(state: TextProcessState) - TextProcessState: print(清洗前State, state.raw_text) # 清洗逻辑... print(清洗后State, state.cleaned_text) return state二节点执行耗时分析添加耗时统计在节点函数中记录执行时间识别低效节点python运行import time def summarize_text_node(state: TextProcessState) - TextProcessState: start_time time.time() # 摘要逻辑... end_time time.time() print(f摘要节点执行耗时{end_time - start_time:.2f}s) return state三错误处理与容错节点内异常捕获在节点函数中添加try-except避免单个节点报错导致整个工作流中断python运行def process_image_node(state: DataProcessState) - DataProcessState: try: # 图像处理逻辑... state.result f图像处理完成{state.data[6:]} except Exception as e: state.result f图像处理失败{str(e)} return state五、常见问题与避坑指南问题现象核心原因解决方案节点函数执行后 State 字段未更新忘记修改 State 或未返回 State 实例确保节点函数中修改 State 字段后返回该实例return state条件边分支判断失效判断函数返回值与mapping中的键不匹配检查返回值类型需为字符串与键名一致性避免拼写错误工作流执行提示 “节点 ID 不存在”添加边时使用了未定义的节点 ID确保add_edge/add_conditional_edges中的节点 ID 已通过add_node注册状态字段类型错误初始 State 传入的字段类型与定义不一致严格按照 State 类的类型注解传入数据如raw_text需为字符串异步节点执行卡顿未使用异步函数或未等待异步操作异步节点需定义为async def并在执行时使用await app.ainvoke()