langgraph笔记

文章目录机制核心代码是哪些?基础示例1基础示例2-加入了大模型为什么叫边不叫线?相比于langchain定制程度更高按步骤来的。机制1、初始化 #builder StateGraph(AgentState)2、添加点3、添加边4、编译5、运行 #app.stream()核心代码是哪些?AgentState相当于一个共享对象。classAgentState(TypedDict):messages:Annotated[List[str],operator.add]final_answer:str加入点和边(edge)这样就形成了图。builderStateGraph(AgentState)# 添加节点builder.add_node(ai_engine,llm_node)builder.add_node(human_gate,human_review)# 定义流程开始 - AI - 人工 - 结束builder.add_edge(START,ai_engine)builder.add_edge(ai_engine,human_gate)builder.add_edge(human_gate,END)appbuilder.compile()基础示例1代码importoperatorfromtypingimportAnnotated,List,TypedDictfromlanggraph.graphimportStateGraph,START,END# # 1. 定义状态 (State)# # 这是整个工作流的“记忆”所有节点都共享这个状态classAgentState(TypedDict):messages:Annotated[List[str],operator.add]# 消息列表operator.add 表示新消息会追加到列表translation:str# 存放翻译结果# # 2. 定义节点 (Nodes)# defai_translate(state:AgentState):模拟 AI 翻译节点print(f AI 正在思考:{state[messages][-1]}...)# 这里模拟一个简单的翻译逻辑实际项目中你会在这里调用 LLMlast_msgstate[messages][-1]if你好inlast_msg:translatedHello, how are you doing?else:translatedfTranslation of:{last_msg}print(f AI 翻译完成:{translated})# 更新状态return{messages:[AI 已生成翻译],translation:translated}defhuman_review(state:AgentState):模拟人工审核节点print(\n--- ⏸️ 暂停等待人工审核 ---)print(f当前翻译结果是【{state[translation]}】)# 在实际应用中这里会挂起等待 API 回调# 这里为了演示我们直接模拟用户说 通过print( 人类看起来没问题通过(模拟自动通过))return{messages:[人类已审核通过]}# # 3. 构建图 (Build Graph)# # 初始化构建器builderStateGraph(AgentState)# 添加节点builder.add_node(translator,ai_translate)builder.add_node(reviewer,human_review)# 定义边流程控制# 流程开始 - 翻译 - 审核 - 结束builder.add_edge(START,translator)builder.add_edge(translator,reviewer)builder.add_edge(reviewer,END)# 编译成可执行的图appbuilder.compile()# # 4. 运行# if__name____main__:print( 启动 LangGraph 工作流...\n)# 初始输入inputs{messages:[你好],translation:}# 调用 invoke 开始运行# stream 模式可以看到每一步的状态变化foroutputinapp.stream(inputs):# output 是一个字典key 是节点名value 是该节点返回的状态更新forkey,valueinoutput.items():print(f✅ 节点 [{key}] 执行完毕返回:{value})print(\n 流程结束)print(f最终结果:{value.get(translation)})基础示例2-加入了大模型代码importtorchfromtransformersimportAutoModelForCausalLM,AutoTokenizerfromlanggraph.graphimportStateGraph,START,ENDfromtypingimportAnnotated,List,TypedDictimportoperator# # 1. 加载本地模型 (Qwen2.5-0.5B)# print(⏳ 正在加载本地模型 (首次运行需下载)...)model_nameQwen/Qwen2.5-0.5B-Instruct# 自动检测是否有 GPUdevicecudaiftorch.cuda.is_available()elsecpuprint(f 使用设备:{device})tokenizerAutoTokenizer.from_pretrained(model_name)modelAutoModelForCausalLM.from_pretrained(model_name,torch_dtypetorch.float16ifdevicecudaelsetorch.float32,# 显卡用半精度CPU用单精度device_mapauto)# # 2. 定义 LangGraph 状态# classAgentState(TypedDict):messages:Annotated[List[str],operator.add]final_answer:str# # 3. 定义节点# defllm_node(state:AgentState):调用本地 Qwen 模型进行推理user_inputstate[messages][-1]print(f [LLM] 收到输入:{user_input})# 构造 Qwen 的对话提示词格式messages[{role:system,content:你是一个聪明的助手请简短回答问题。},{role:user,content:user_input}]# 应用聊天模板texttokenizer.apply_chat_template(messages,tokenizeFalse,add_generation_promptTrue)model_inputstokenizer([text],return_tensorspt).to(device)# 生成回答generated_idsmodel.generate(model_inputs.input_ids,max_new_tokens512,do_sampleTrue,temperature0.7)# 解码输出generated_ids[output_ids[len(input_ids):]forinput_ids,output_idsinzip(model_inputs.input_ids,generated_ids)]responsetokenizer.batch_decode(generated_ids,skip_special_tokensTrue)[0]print(f [LLM] 回答:{response})return{messages:[AI已回复],final_answer:response}defhuman_review(state:AgentState):人工审核节点print(\n--- ⏸️ 暂停等待人工审核 ---)print(f AI 生成的回答是【{state[final_answer]}】)# 模拟人工确认confirminput( 是否满意(输入 y 继续其他退出): )ifconfirm.lower()y:return{messages:[用户确认通过]}else:print(❌ 用户拒绝流程终止。)# 这里简单处理实际可以跳转到重写节点exit()# # 4. 构建并运行图# builderStateGraph(AgentState)# 添加节点builder.add_node(ai_engine,llm_node)builder.add_node(human_gate,human_review)# 定义流程开始 - AI - 人工 - 结束builder.add_edge(START,ai_engine)builder.add_edge(ai_engine,human_gate)builder.add_edge(human_gate,END)appbuilder.compile()if__name____main__:print(\n LangGraph Qwen 本地智能体启动 \n)whileTrue:queryinput( 请输入问题 (输入 quit 退出): )ifqueryquit:breakinputs{messages:[query],final_answer:}# 运行工作流foroutputinapp.stream(inputs):fornode_name,valuesinoutput.items():ifvalues.get(final_answer):print(f✅ 最终结果:{values[final_answer]})print(-*30)为什么叫边不叫线?点和边是图中两个核心概念。之所以不叫线是因为线是没有方向的但是边有方向(图概念里面的边和平行四边形的边可不一样)。