把大模型“画”成图：LangGraph 超详细入门 + 实战

创想小编 · 发表于 4 天前

作者：CSDN博客
0. 为什么需要 LangGraph？

传统 LangChain 用“链”描述线性流程，一到循环、分支、并行、回溯就捉襟见肘。
LangGraph 把整条链路升级成有向图：

结果：多轮对话、工具调用、循环反思都能一张图表达，代码量反而更少。

1. 核心概念 4 件套（5 分钟速记）

组件	解释	类比
State	全局字典，节点间传递数据	内存条
Node	处理函数，接收 State → 返回新 State	CPU 指令
Edge	连接节点，可条件分支	总线
Graph	整张有向图，编译后可调用	主板电路

所有节点只读写同一份 State，天然解耦，又自带“记忆”。

2. 安装 & 最小可运行 Demo

pip install langgraph langchain-core

复制代码

from typing import TypedDict, List
from langgraph.graph import StateGraph, START, END
# ① 定义 StateclassAgentState(TypedDict):
messages: List[str]
current_step:str# ② 写节点函数defhello_node(state: AgentState)-> AgentState:
state["messages"].append("Hello from LangGraph!")
state["current_step"]="hello"return state
defbye_node(state: AgentState)-> AgentState:
state["messages"].append("Bye ~")
state["current_step"]="bye"return state
# ③ 建图并连边
builder = StateGraph(AgentState)
builder.add_node("hello", hello_node)
builder.add_node("bye", bye_node)
builder.add_edge(START,"hello")
builder.add_edge("hello","bye")
builder.add_edge("bye", END)
graph = builder.compile()# ④ 运行
result = graph.invoke({"messages":[],"current_step":""})print(result["messages"])# 输出：['Hello from LangGraph!', 'Bye ~']

复制代码

这就是最小图：START → hello → bye → END。

3. 实战：带条件分支的“智能客服”

场景

用户输入 → 识别意图 → 根据意图

检索知识库

生成工单

直接回答

3.1 定义共享 State

from typing import TypedDict, List, Optional
classState(TypedDict):
user_input:str
intent: Optional[str]=None# 意图
answer: Optional[str]=None# 最终回答
ticket_id: Optional[str]=None# 工单号

复制代码

3.2 实现节点函数

from langchain_community.chat_models import ChatOpenAI
llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0)defclassify_intent(state: State)-> State:
prompt =f"用户输入：{state['user_input']}\n\n请返回意图类别：咨询/投诉/其他"
state["intent"]= llm.invoke(prompt).content.strip()return state
defhandle_consult(state: State)-> State:# 模拟 RAG
state["answer"]="根据知识库，咨询答案是 ..."return state
defhandle_complaint(state: State)-> State:# 模拟生成工单
state["ticket_id"]="T10086"
state["answer"]="已为您生成投诉工单，编号 T10086，专员将在 24h 内联系您。"return state
defhandle_other(state: State)-> State:
state["answer"]= llm.invoke(state["user_input"]).content
return state

复制代码

3.3 条件边：让 LLM 决定走向

defroute_intent(state: State)->str:return state["intent"].lower()

复制代码

3.4 建图 & 编译

from langgraph.graph import StateGraph, START, END
workflow = StateGraph(State)
workflow.add_node("classify", classify_intent)
workflow.add_node("consult", handle_consult)
workflow.add_node("complaint", handle_complaint)
workflow.add_node("other", handle_other)
workflow.add_edge(START,"classify")
workflow.add_conditional_edges("classify",
route_intent,{"咨询":"consult","投诉":"complaint","其他":"other"})
workflow.add_edge("consult", END)
workflow.add_edge("complaint", END)
workflow.add_edge("other", END)
app = workflow.compile()

复制代码

3.5 运行

out = app.invoke({"user_input":"我要投诉你们产品！"})print(out["answer"])# 已为您生成投诉工单，编号 T10086...

复制代码

意图分类 → 条件分支 → 不同处理，一张图解决。

4. 高级技巧：循环 + 反思（Self-Correct）

场景

LLM 先写代码 → 自动执行 → 失败则 反思+重试（最多 3 次）。

import subprocess, re
from langgraph.graph import StateGraph, START, END
classCodeState(TypedDict):
task:str
code:str
error:str|None
attempts:intdefwrite_code(state: CodeState)-> CodeState:
state["code"]= llm.invoke(f"用 Python 实现：{state['task']}，只返回代码块").content
state["attempts"]+=1return state
defrun_code(state: CodeState)-> CodeState:try:
out = subprocess.check_output(["python","-c", state["code"]], stderr=subprocess.STDOUT, text=True)
state["error"]=Noneexcept subprocess.CalledProcessError as e:
state["error"]= e.output
return state
defshould_continue(state: CodeState)->str:if state["error"]isNoneor state["attempts"]>=3:return"end"return"reflect"defreflect(state: CodeState)-> CodeState:
prompt =f"代码执行报错：{state['error']}\n\n请修正代码，只返回修正后的完整代码块。"
state["code"]= llm.invoke(prompt).content
return state
# 建图
g = StateGraph(CodeState)
g.add_node("write", write_code)
g.add_node("run", run_code)
g.add_node("reflect", reflect)
g.add_edge(START,"write")
g.add_edge("write","run")
g.add_conditional_edges("run", should_continue,{"reflect":"reflect","end": END})
g.add_edge("reflect","run")
g.add_edge("run", END, when=lambda s: s["error"]isNone)
bot = g.compile()
final = bot.invoke({"task":"快速排序","code":"","error":None,"attempts":0})print(final["code"])

复制代码

节点 run → 条件边判断 → reflect 节点循环，天然 while 语法。

5. 彩蛋：条件边返回 "human" → 弹出审批框

场景

高风险操作（转账 > 10k、删除数据库）必须经过人工审批才能继续。
5.1 定义状态

classPaymentState(TypedDict):
amount:float
currency:str
to_account:str
approved:bool|None# None=待审批，True=通过，False=拒绝

复制代码

5.2 节点函数

defcheck_amount(state: PaymentState)-> PaymentState:if state["amount"]>10_000:
state["approved"]=None# 标记需审批else:
state["approved"]=True# 小额自动通过return state
defwait_human(state: PaymentState)-> PaymentState:# 这里只是占位，真正部署时会阻塞等待外部审批系统print(f"[系统] 等待人工审批：向 {state['to_account']} 转账 {state['amount']}{state['currency']}")# 模拟审批结果（实际从外部系统读取）
state["approved"]=input("审批结果（y/n）：").lower()=="y"return state
defexecute(state: PaymentState)-> PaymentState:print(f"[执行] 转账成功 ✅")return state
defreject(state: PaymentState)-> PaymentState:print(f"[拒绝] 转账已取消 ❌")return state

复制代码

5.3 条件边：返回 "human" 或 "exec" 或 "reject"

defroute_after_check(state: PaymentState)->str:if state["approved"]isNone:return"human"# 需要人工审批return"exec"if state["approved"]else"reject"

复制代码

5.4 建图 & 编译

from langgraph.graph import StateGraph, START, END
workflow = StateGraph(PaymentState)
workflow.add_node("check", check_amount)
workflow.add_node("human", wait_human)# 人工审批节点
workflow.add_node("exec", execute)
workflow.add_node("reject", reject)
workflow.add_edge(START,"check")
workflow.add_conditional_edges("check", route_after_check,{"human":"human","exec":"exec","reject":"reject"})
workflow.add_edge("human","exec")# 审批通过后执行
workflow.add_edge("human","reject", when=lambda s: s["approved"]isFalse)
workflow.add_edge("exec", END)
workflow.add_edge("reject", END)
app = workflow.compile()

复制代码

5.5 运行：大额触发审批

# 小额自动通过
app.invoke({"amount":100,"currency":"CNY","to_account":"A123","approved":None})# 输出：[执行] 转账成功 ✅# 大额等待人工输入
app.invoke({"amount":50000,"currency":"CNY","to_account":"B456","approved":None})# 控制台弹出：# [系统] 等待人工审批：向 B456 转账 50000.0 CNY# 审批结果（y/n）：y# [执行] 转账成功 ✅

复制代码

5.6 生产级扩展

前端弹窗

REST 阻塞调用

超时拒绝

审计日志

LangSmith / 自建审计表

6. 与 LangChain 无缝集成

任何 langchain 的 LLM、Tool、Retriever 都能直接当节点用：

from langchain.tools import ShellTool
from langchain import hub
from langgraph.graph import StateGraph
prompt = hub.pull("hwchase17/react")# 官方 ReAct prompt
tool = ShellTool()
agent_executor = create_react_agent(model,[tool], prompt)# LangGraph 内置
graph = agent_executor # 本身就是一张图
graph.invoke({"input":"列出当前目录下文件并按大小排序"})

复制代码

一行代码把传统 AgentExecutor 升级成有状态图。

7. 可视化调试：随手出图

from IPython.display import Image
Image(app.get_graph().draw_mermaid_png())

复制代码

自动生成 Mermaid 流程图，节点-边-条件一目了然，调试效率 ×3。

8. 生产级 Tips

断点续跑

重启后从断点继续

人机协同

关键操作双保险

并行化

同时生成故事+翻译+摘要

9. 总结：一张图记住 LangGraph

State（共享内存）
↑
Node（函数/LLM/Tool） ←→ Edge（条件或顺序）
↓
START / END / 循环 / 并行 / 人机

复制代码

把业务画成图，把图跑成代码——LangGraph 让复杂代理从“玄学”变“工程”。

下一步

循环优化查询

多智能体

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

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

25位大厂高管转战生成式AI创业！吸金猛，不

把大模型“画”成图：LangGraph 超详细入门 + 实战

一文讲清智能体（AI Agent），这是一篇不得

关于我们

产品与服务

解决方案

产品与服务