一条“速成路线”把 LangGraph 讲清楚
LangGraph是一个基于状态图的智能体编排框架,支持循环、可中断和持久化的工作流,内置检查点和人工参与机制。学习路径:1)理解核心概念;2)通过预置ReAct智能体快速上手;3)掌握StateGraph等底层原语;4)实现持久化和人工干预;5)使用Studio调试和云端部署。提供两个示例:A)预置ReAct智能体调用工具;B)自定义节点+SQLite持久化+人工中断。注意包名变更、检查点依赖等
一句话解释
LangGraph 是一个“有状态(stateful)的智能体编排框架”,用“节点-边-状态图”来搭建可循环、可中断、可持久化的智能体/多智能体工作流;内置检查点(checkpoint)与人参与(HITL)机制,支持 Python/JS,并提供 Studio 与云端平台用于可视化调试和部署。
⸻
学习 LangGraph 的最短路径(半天到一天)
① 先扫概念:理解为什么选“图”来组织智能体:可控、可循环、可插人审。(关键词:State、Channels、Nodes、Edges、条件路由、循环、持久化、HITL)
② 5分钟起步:用预置 Agent(ReAct 模式)
安装后直接用 create_react_agent 把“模型 + 工具”包成一个能循环调工具的智能体。自 v0.3 起预置 Agent 拆分为 langgraph-prebuilt 包。
③ 低阶原语:自己用 StateGraph 拼节点、定义状态与 reducer(如 add_messages),掌控每一步执行与路由。
④ 持久化与人参与:给图编译上 checkpointer(如 SQLite/Postgres),就能获得“线程(thread)/历史/回放/分叉/断点继续”,并可在节点里 interrupt() 暂停等待人工确认再继续。注:SQLite/Postgres 的 checkpointer 需额外安装对应包。
⑤ 调试与部署:
• 本地:用 LangGraph Studio 可视化运行、打断点;langgraph dev 一键起本地 Server。
• 线上:用 LangGraph Platform/Server 托管,有线程、任务队列与 API(Assistants、Threads、Runs、Webhooks 等)。
⸻
可直接跑的两个最小示例
A. 预置 ReAct Agent(最快上手)
pip install -U langgraph langgraph-prebuilt langchain-openai
需要 OPENAI_API_KEY(或改用你常用的 LLM 提供方)
from langchain_core.tools import tool
from langchain_openai import ChatOpenAI
from langgraph.prebuilt import create_react_agent
@tool
def calc(x: int, y: int) -> int:
"""演示用的小工具:加法"""
return x + y
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
agent = create_react_agent(llm, tools=[calc])
result = agent.invoke({
"messages": [("user", "用工具算一下 7+35,然后用中文解释下过程")]
})
print(result["messages"][-1].content)
(create_react_agent 会在“LLM ↔ 工具调用”之间循环,直到满足停止条件。)
B. 低阶原语 + 持久化 + 可中断
pip install -U langgraph langgraph-checkpoint-sqlite langchain-openai
from typing import Annotated, TypedDict
from langgraph.graph import StateGraph, END
from langgraph.graph.message import add_messages
from langchain_openai import ChatOpenAI
from langgraph.checkpoint.sqlite import SqliteSaver
from langgraph.types import interrupt
class State(TypedDict):
messages: Annotated[list, add_messages]
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
def assistant(state: State):
# 在关键点请求人工确认(HITL)
need_ok = interrupt({"ask": "是否继续调用模型?(y/n)"})
if isinstance(need_ok, dict) and need_ok.get("answer","y").lower() != "y":
return {"messages": [("assistant", "已取消")]}
msg = llm.invoke(state["messages"])
return {"messages": [msg]}
builder = StateGraph(State)
builder.add_node("assistant", assistant)
builder.set_entry_point("assistant")
builder.add_edge("assistant", END)
checkpointer = SqliteSaver.from_conn_string(":memory:")
app = builder.compile(checkpointer=checkpointer)
config = {"configurable": {"thread_id": "demo-1"}}
第一次调用会在 interrupt 处暂停,返回 interrupt,等待你用 Command.resume 继续
first = app.invoke({"messages": [("user","你好!先别急,等我确认")]}, config)
print(first.get("__interrupt__"))
……收集到人工输入后继续(伪代码):
from langgraph.types import Command
for update in app.stream(Command(resume={"answer": "y"}), config=config):
pass
(要点:① 安装 SQLite checkpointer 才能持久化;② interrupt() 依赖持久化来记录图状态并恢复执行。)
⸻
常见坑位(避坑清单)
• 预置 Agent 的包名变化:自 v0.3 起,预置 Agent(含 create_react_agent)迁到 langgraph-prebuilt;老教程若导入失败,请安装新包或按文档更新。
• Checkpointer 需单独安装:langgraph-checkpoint-sqlite / -postgres 分别提供 SqliteSaver/PostgresSaver;不装就没有持久化与 HITL。
• 本地 Server/CLI 要求:用 langgraph dev 起本地服务与 Studio 调试时,官方文档标注 Python ≥ 3.11。
⸻
进阶一路通
• 多智能体范式:Supervisor/Swarm 等模板可直接复用,然后再下潜到低阶原语做深改。
• 平台化部署:LangGraph Server/Platform 提供 Assistants/Threads/Runs 的 API 以及队列与持久化,适合把你的图“变成服务”。
• 与 LangChain 的关系:LangChain 的 Agents 现已建立在 LangGraph 之上,借用其“持久化、HITL、可恢复执行”等能力。
更多推荐
所有评论(0)