langgraph笔记

机制

1、初始化 #builder = StateGraph(AgentState)
2、添加点
3、添加边
4、编译
5、运行 #app.stream()

核心代码是哪些?

AgentState相当于一个共享对象。

classAgentState(TypedDict):messages:Annotated[List[str],operator.add]final_answer:str

加入点和边(edge)，这样就形成了图。

builder=StateGraph(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)app=builder.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]}...")# 这里模拟一个简单的翻译逻辑（实际项目中你会在这里调用 LLM）last_msg=state['messages'][-1]if"你好"inlast_msg:translated="Hello, how are you doing?"else:translated=f"Translation 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)# ==========================================# 初始化构建器builder=StateGraph(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)# 编译成可执行的图app=builder.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_name="Qwen/Qwen2.5-0.5B-Instruct"# 自动检测是否有 GPUdevice="cuda"iftorch.cuda.is_available()else"cpu"print(f"🚀 使用设备:{device}")tokenizer=AutoTokenizer.from_pretrained(model_name)model=AutoModelForCausalLM.from_pretrained(model_name,torch_dtype=torch.float16ifdevice=="cuda"elsetorch.float32,# 显卡用半精度，CPU用单精度device_map="auto")# ==========================================# 2. 定义 LangGraph 状态# ==========================================classAgentState(TypedDict):messages:Annotated[List[str],operator.add]final_answer:str# ==========================================# 3. 定义节点# ==========================================defllm_node(state:AgentState):"""调用本地 Qwen 模型进行推理"""user_input=state['messages'][-1]print(f"🤖 [LLM] 收到输入:{user_input}")# 构造 Qwen 的对话提示词格式messages=[{"role":"system","content":"你是一个聪明的助手，请简短回答问题。"},{"role":"user","content":user_input}]# 应用聊天模板text=tokenizer.apply_chat_template(messages,tokenize=False,add_generation_prompt=True)model_inputs=tokenizer([text],return_tensors="pt").to(device)# 生成回答generated_ids=model.generate(model_inputs.input_ids,max_new_tokens=512,do_sample=True,temperature=0.7)# 解码输出generated_ids=[output_ids[len(input_ids):]forinput_ids,output_idsinzip(model_inputs.input_ids,generated_ids)]response=tokenizer.batch_decode(generated_ids,skip_special_tokens=True)[0]print(f"🤖 [LLM] 回答:{response}")return{"messages":["AI已回复"],"final_answer":response}defhuman_review(state:AgentState):"""人工审核节点"""print("\n--- ⏸️ 暂停：等待人工审核 ---")print(f"👀 AI 生成的回答是：【{state['final_answer']}】")# 模拟人工确认confirm=input("👤 是否满意？(输入 y 继续，其他退出): ")ifconfirm.lower()=='y':return{"messages":["用户确认通过"]}else:print("❌ 用户拒绝，流程终止。")# 这里简单处理，实际可以跳转到重写节点exit()# ==========================================# 4. 构建并运行图# ==========================================builder=StateGraph(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)app=builder.compile()if__name__=="__main__":print("\n🚀 === LangGraph + Qwen 本地智能体启动 ===\n")whileTrue:query=input("👤 请输入问题 (输入 'quit' 退出): ")ifquery=='quit':breakinputs={"messages":[query],"final_answer":""}# 运行工作流foroutputinapp.stream(inputs):fornode_name,valuesinoutput.items():ifvalues.get('final_answer'):print(f"✅ 最终结果:{values['final_answer']}")print("-"*30)

为什么叫边不叫线?

点和边是图中两个核心概念。
之所以不叫线，是因为线是没有方向的，但是边有方向(图概念里面的边和平行四边形的边可不一样)。