AI创想

标题: 把大模型“画”成图：LangGraph 超详细入门 + 实战 [打印本页]
作者: 创想小编    时间: 4 天前
标题: 把大模型“画”成图：LangGraph 超详细入门 + 实战
作者：CSDN博客
0. 为什么需要 LangGraph？

传统 LangChain 用“链”描述线性流程，一到循环、分支、并行、回溯就捉襟见肘。
LangGraph 把整条链路升级成有向图：
结果：多轮对话、工具调用、循环反思都能一张图表达，代码量反而更少。
1. 核心概念 4 件套（5 分钟速记）

组件解释类比
State全局字典，节点间传递数据内存条
Node处理函数，接收 State → 返回新 StateCPU 指令
Edge连接节点，可条件分支总线
Graph整张有向图，编译后可调用主板电路
所有节点只读写同一份 State，天然解耦，又自带“记忆”。
2. 安装 & 最小可运行 Demo
  1. pip install langgraph langchain-core
复制代码
  1. from typing import TypedDict, List
  2. from langgraph.graph import StateGraph, START, END
  4. # ① 定义 StateclassAgentState(TypedDict):
  5.     messages: List[str]
  6.     current_step:str# ② 写节点函数defhello_node(state: AgentState)-> AgentState:
  7.     state["messages"].append("Hello from LangGraph!")
  8.     state["current_step"]="hello"return state
  10. defbye_node(state: AgentState)-> AgentState:
  11.     state["messages"].append("Bye ~")
  12.     state["current_step"]="bye"return state
  14. # ③ 建图并连边
  15. builder = StateGraph(AgentState)
  16. builder.add_node("hello", hello_node)
  17. builder.add_node("bye",  bye_node)
  18. builder.add_edge(START,"hello")
  19. builder.add_edge("hello","bye")
  20. builder.add_edge("bye",  END)
  22. graph = builder.compile()# ④ 运行
  23. result = graph.invoke({"messages":[],"current_step":""})print(result["messages"])# 输出：['Hello from LangGraph!', 'Bye ~']
复制代码
这就是最小图：START → hello → bye → END。
3. 实战：带条件分支的“智能客服”

场景

用户输入 → 识别意图 → 根据意图
3.1 定义共享 State
  1. from typing import TypedDict, List, Optional
  3. classState(TypedDict):
  4.     user_input:str
  5.     intent: Optional[str]=None# 意图
  6.     answer: Optional[str]=None# 最终回答
  7.     ticket_id: Optional[str]=None# 工单号
复制代码
3.2 实现节点函数
  1. from langchain_community.chat_models import ChatOpenAI
  3. llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0)defclassify_intent(state: State)-> State:
  4.     prompt =f"用户输入：{state['user_input']}\n\n请返回意图类别：咨询/投诉/其他"
  5.     state["intent"]= llm.invoke(prompt).content.strip()return state
  7. defhandle_consult(state: State)-> State:# 模拟 RAG
  8.     state["answer"]="根据知识库，咨询答案是 ..."return state
  10. defhandle_complaint(state: State)-> State:# 模拟生成工单
  11.     state["ticket_id"]="T10086"
  12.     state["answer"]="已为您生成投诉工单，编号 T10086，专员将在 24h 内联系您。"return state
  14. defhandle_other(state: State)-> State:
  15.     state["answer"]= llm.invoke(state["user_input"]).content
  16.     return state
复制代码
3.3 条件边：让 LLM 决定走向
  1. defroute_intent(state: State)->str:return state["intent"].lower()
复制代码
3.4 建图 & 编译
  1. from langgraph.graph import StateGraph, START, END
  3. workflow = StateGraph(State)
  4. workflow.add_node("classify", classify_intent)
  5. workflow.add_node("consult",  handle_consult)
  6. workflow.add_node("complaint", handle_complaint)
  7. workflow.add_node("other",     handle_other)
  9. workflow.add_edge(START,"classify")
  10. workflow.add_conditional_edges("classify",
  11.     route_intent,{"咨询":"consult","投诉":"complaint","其他":"other"})
  12. workflow.add_edge("consult",  END)
  13. workflow.add_edge("complaint", END)
  14. workflow.add_edge("other",    END)
  16. app = workflow.compile()
复制代码
3.5 运行
  1. out = app.invoke({"user_input":"我要投诉你们产品！"})print(out["answer"])# 已为您生成投诉工单，编号 T10086...
复制代码
意图分类 → 条件分支 → 不同处理，一张图解决。
4. 高级技巧：循环 + 反思（Self-Correct）

场景

LLM 先写代码 → 自动执行 → 失败则 反思+重试（最多 3 次）。
  1. import subprocess, re
  2. from langgraph.graph import StateGraph, START, END
  4. classCodeState(TypedDict):
  5.     task:str
  6.     code:str
  7.     error:str|None
  8.     attempts:intdefwrite_code(state: CodeState)-> CodeState:
  9.     state["code"]= llm.invoke(f"用 Python 实现：{state['task']}，只返回代码块").content
  10.     state["attempts"]+=1return state
  12. defrun_code(state: CodeState)-> CodeState:try:
  13.         out = subprocess.check_output(["python","-c", state["code"]], stderr=subprocess.STDOUT, text=True)
  14.         state["error"]=Noneexcept subprocess.CalledProcessError as e:
  15.         state["error"]= e.output
  16.     return state
  18. defshould_continue(state: CodeState)->str:if state["error"]isNoneor state["attempts"]>=3:return"end"return"reflect"defreflect(state: CodeState)-> CodeState:
  19.     prompt =f"代码执行报错：{state['error']}\n\n请修正代码，只返回修正后的完整代码块。"
  20.     state["code"]= llm.invoke(prompt).content
  21.     return state
  23. # 建图
  24. g = StateGraph(CodeState)
  25. g.add_node("write",  write_code)
  26. g.add_node("run",   run_code)
  27. g.add_node("reflect", reflect)
  28. g.add_edge(START,"write")
  29. g.add_edge("write","run")
  30. g.add_conditional_edges("run", should_continue,{"reflect":"reflect","end": END})
  31. g.add_edge("reflect","run")
  32. g.add_edge("run", END, when=lambda s: s["error"]isNone)
  34. bot = g.compile()
  35. final = bot.invoke({"task":"快速排序","code":"","error":None,"attempts":0})print(final["code"])
复制代码
节点 run → 条件边判断 → reflect 节点循环，天然 while 语法。
5. 彩蛋：条件边返回 "human" → 弹出审批框

场景

高风险操作（转账 > 10k、删除数据库）必须经过人工审批才能继续。
5.1 定义状态
  1. classPaymentState(TypedDict):
  2.     amount:float
  3.     currency:str
  4.     to_account:str
  5.     approved:bool|None# None=待审批，True=通过，False=拒绝
复制代码
5.2 节点函数
  1. defcheck_amount(state: PaymentState)-> PaymentState:if state["amount"]>10_000:
  2.         state["approved"]=None# 标记需审批else:
  3.         state["approved"]=True# 小额自动通过return state
  5. defwait_human(state: PaymentState)-> PaymentState:# 这里只是占位，真正部署时会阻塞等待外部审批系统print(f"[系统] 等待人工审批：向 {state['to_account']} 转账 {state['amount']}{state['currency']}")# 模拟审批结果（实际从外部系统读取）
  6.     state["approved"]=input("审批结果（y/n）：").lower()=="y"return state
  8. defexecute(state: PaymentState)-> PaymentState:print(f"[执行] 转账成功 ✅")return state
  10. defreject(state: PaymentState)-> PaymentState:print(f"[拒绝] 转账已取消 ❌")return state
复制代码
5.3 条件边：返回 "human" 或 "exec" 或 "reject"
  1. defroute_after_check(state: PaymentState)->str:if state["approved"]isNone:return"human"# 需要人工审批return"exec"if state["approved"]else"reject"
复制代码
5.4 建图 & 编译
  1. from langgraph.graph import StateGraph, START, END
  3. workflow = StateGraph(PaymentState)
  4. workflow.add_node("check",   check_amount)
  5. workflow.add_node("human",   wait_human)# 人工审批节点
  6. workflow.add_node("exec",    execute)
  7. workflow.add_node("reject",  reject)
  9. workflow.add_edge(START,"check")
  10. workflow.add_conditional_edges("check", route_after_check,{"human":"human","exec":"exec","reject":"reject"})
  11. workflow.add_edge("human","exec")# 审批通过后执行
  12. workflow.add_edge("human","reject", when=lambda s: s["approved"]isFalse)
  13. workflow.add_edge("exec", END)
  14. workflow.add_edge("reject", END)
  16. app = workflow.compile()
复制代码
5.5 运行：大额触发审批
  1. # 小额自动通过
  2. app.invoke({"amount":100,"currency":"CNY","to_account":"A123","approved":None})# 输出：[执行] 转账成功 ✅# 大额等待人工输入
  3. app.invoke({"amount":50000,"currency":"CNY","to_account":"B456","approved":None})# 控制台弹出：# [系统] 等待人工审批：向 B456 转账 50000.0 CNY# 审批结果（y/n）：y# [执行] 转账成功 ✅
复制代码
5.6 生产级扩展

6. 与 LangChain 无缝集成

任何 langchain 的 LLM、Tool、Retriever 都能直接当节点用：
  1. from langchain.tools import ShellTool
  2. from langchain import hub
  3. from langgraph.graph import StateGraph
  5. prompt = hub.pull("hwchase17/react")# 官方 ReAct prompt
  6. tool = ShellTool()
  7. agent_executor = create_react_agent(model,[tool], prompt)# LangGraph 内置
  9. graph = agent_executor  # 本身就是一张图
  10. graph.invoke({"input":"列出当前目录下文件并按大小排序"})
复制代码
一行代码把传统 AgentExecutor 升级成有状态图。
7. 可视化调试：随手出图
  1. from IPython.display import Image
  2. Image(app.get_graph().draw_mermaid_png())
复制代码
自动生成 Mermaid 流程图，节点-边-条件一目了然，调试效率 ×3。
8. 生产级 Tips

9. 总结：一张图记住 LangGraph
  1. State（共享内存）
  2.    ↑
  3. Node（函数/LLM/Tool） ←→ Edge（条件或顺序）
  4.    ↓
  5. START / END / 循环 / 并行 / 人机
复制代码
把业务画成图，把图跑成代码——LangGraph 让复杂代理从“玄学”变“工程”。
下一步

现在就 pip install langgraph，把今天的示例跑通，你就拥有了第一张“会循环的 AI 工作流”！
学习代码

原文地址：https://blog.csdn.net/axibazZ/article/details/156677608



欢迎光临 AI创想 (https://llms-ai.com/) Powered by Discuz! X3.4