Loading...

文章背景图

LangGraph:有状态Agent图编排框架安装与使用指南

2026-10-24
0
- 字
- 分钟
|

LangGraph

LangGraph:有状态Agent图编排框架安装与使用指南

项目名称langchain-ai/langgraph
仓库地址https://github.com/langchain-ai/langgraph
官方网站https://www.langchain.com/langgraph
官方文档LangGraph Docs
最新更新时间2026-10-24(持续活跃更新)
项目定位LangChain 团队推出的有状态多 Actor Agent 图编排框架,支持循环、分支、持久化
开源协议MIT

一、项目背景

LangChain 在 2023 年火遍全网后,很快暴露了一个问题:它的 Chain 是"DAG"——有向无环图,只能 A→B→C 一次性跑到底。但真实的 Agent 系统需要"循环"——让 LLM 反复思考、调工具、观察结果、再思考,直到任务完成;还需要"分支"——根据中间结果决定下一步走哪条路;更需要"持久化"——用户关掉网页再打开,对话历史和中间状态还在。

LangGraph 由 LangChain 团队在 2024 年初推出,把 Agent 抽象成一张"有状态图":节点是函数(调用 LLM、调工具、做检索),边是状态转移(普通边、条件边),全局 State 是一个 TypedDict,在节点之间流动。这种范式非常接近有限状态机(FSM),但比传统 FSM 更灵活——节点内部可以是任意 LLM 调用。

LangGraph 已经成为 LangChain 生态里生产级 Agent 的首选方案:LangChain 官方的 Deep Research、ReAct、Plan-and-Execute 模板都基于它;大量初创公司用它做客服 Agent、代码 Agent、研究助手。它还配套了 LangGraph Platform(云托管)和 LangGraph Studio(可视化调试 UI),形成完整工具链。

二、核心特性

  • 显式状态机:用 StateGraph 定义节点与边,流程完全可控,不是黑盒。
  • 循环与条件分支:支持 cycle(Agent 反复思考-行动)、conditional edges(根据状态走不同分支)。
  • 持久化 Checkpoint:内置 SQLite/Postgres/Redis 检查点,支持中断恢复、时间旅行调试。
  • Human-in-the-loop:节点可暂停等待人类审批,批准后继续执行。
  • 多 Agent 协作:支持 Supervisor、Swarm、Network 等多 Agent 拓扑。
  • 流式输出:支持 token-level、节点-level、状态-level 三种流式。
  • 与 LangChain 生态无缝集成:可直接复用 LangChain 的 Tool、ChatModel、Retriever。
  • LangGraph Studio:可视化拖拽调试、单步执行、看状态变化。

三、环境要求

  • Python:3.10+,推荐 3.11。
  • 操作系统:Linux / macOS / WSL2。
  • LLM API Key:OpenAI / Anthropic / DeepSeek 等。
  • (可选)LangSmith:用于追踪调试,注册免费账号即可。
  • 内存:2 GB+。

四、详细部署步骤

第一步:安装

python3 -m venv .venv
source .venv/bin/activate

# 核心包
pip install -U langgraph

# 配合 LangChain 使用
pip install -U langchain-openai langchain-anthropic

# 搜索工具示例
pip install -U tavily-python

第二步:配置 LLM

import os
os.environ["OPENAI_API_KEY"] = "sk-xxxx"
# 国内 DeepSeek:
# os.environ["OPENAI_API_BASE"] = "https://api.deepseek.com/v1"

from langchain_openai import ChatOpenAI
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

第三步:写一个最简单的 Agent

from typing import Annotated
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages

# 1) 定义全局状态
class State(TypedDict):
    messages: Annotated[list, add_messages]   # add_messages 自动拼接历史

# 2) 定义节点
def chatbot(state: State):
    return {"messages": [llm.invoke(state["messages"])]}

# 3) 组装图
graph_builder = StateGraph(State)
graph_builder.add_node("chatbot", chatbot)
graph_builder.add_edge(START, "chatbot")
graph_builder.add_edge("chatbot", END)

graph = graph_builder.compile()

# 4) 跑起来
result = graph.invoke({"messages": [("user", "你好,介绍一下你自己")]})
print(result["messages"][-1].content)

第四步:带工具调用的 ReAct Agent

from langchain_core.tools import tool
from langgraph.prebuilt import ToolNode, tools_condition

@tool
def get_weather(city: str):
    """查询城市天气"""
    return f"{city} 今天晴 22℃"

tools = [get_weather]
llm_with_tools = llm.bind_tools(tools)

def call_model(state: State):
    return {"messages": [llm_with_tools.invoke(state["messages"])]}

builder = StateGraph(State)
builder.add_node("agent", call_model)
builder.add_node("tools", ToolNode(tools))

builder.add_edge(START, "agent")
# 条件边:如果 LLM 要求调工具,去 tools 节点;否则结束
builder.add_conditional_edges("agent", tools_condition)
# 工具执行完回到 agent
builder.add_edge("tools", "agent")

graph = builder.compile()

# 跑
for chunk in graph.stream(
    {"messages": [("user", "北京天气怎么样?")]},
    stream_mode="values",
):
    chunk["messages"][-1].pretty_print()

第五步:加持久化(中断恢复)

from langgraph.checkpoint.sqlite import SqliteSaver

with SqliteSaver.from_conn_string("checkpoints.sqlite") as checkpointer:
    graph = builder.compile(checkpointer=checkpointer)

    config = {"configurable": {"thread_id": "user-001"}}
    graph.invoke({"messages": [("user", "记住我叫若琳")]}, config)

    # 下次再问,历史还在
    graph.invoke({"messages": [("user", "我叫什么?")]}, config)

换成 PostgreSQL 只需把 SqliteSaver 换成 PostgresSaver。

第六步:Human-in-the-loop 审批

from langgraph.types import interrupt, Command

def sensitive_node(state: State):
    # 这里会暂停,等外部 Command(resume=...) 恢复
    user_approval = interrupt({"question": "是否允许执行删除操作?"})
    if user_approval == "yes":
        return {"messages": [("assistant", "已执行删除")]}
    else:
        return {"messages": [("assistant", "已取消")]}

第七步:多 Agent 协作(Supervisor 模式)

from langgraph.graph import StateGraph
# 多个 worker agent + 一个 supervisor,supervisor 决定下一步派给谁
# 官方有现成模板:
# pip install langgraph-supervisor
from langgraph.supervisor import create_supervisor

可视化图结构

from IPython.display import Image, display
display(Image(graph.get_graph().draw_mermaid_png()))

也可以导出 Mermaid 源码:

print(graph.get_graph().draw_mermaid())

五、进阶生产部署

5.1 LangGraph Platform 部署

生产推荐用官方的 LangGraph Server:

pip install langgraph-cli
langgraph build -t my-agent .     # 打成 Docker 镜像
docker run -p 8123:8000 my-agent  # 启动
# 提供 /assistants、/threads、/runs 等 REST API

5.2 监控与调试

  • 接 LangSmith:export LANGSMITH_TRACING=true,每次 graph.run 都有 trace;
  • 用 graph.get_state(config) 查看当前状态;
  • 用 graph.update_state(config, {...}) 人工干预状态。

5.3 性能优化

  • 把大对象(文档、向量)放到外部存储,State 里只放 ID;
  • 长流程用 PostgreSQL Checkpointer 而不是 SQLite;
  • 用 asyncio 并发跑多个无依赖节点。

六、使用方法示例

流式输出(SSE 给前端):

for event in graph.stream({"messages": [...]}, stream_mode="updates"):
    print(event)

LangGraph Server(生产部署):

pip install langgraph-cli
langgraph dev        # 本地起一个带 API 的服务
# 访问 http://localhost:8123 打开 Studio

典型任务示例:

  • 客服机器人:用户提问 → 检索知识库 → 不确定就转人工。
  • 研究助手:定题 → 搜索 → 阅读 → 写大纲 → 写初稿 → 校对。
  • 代码助手:理解 issue → 改代码 → 跑测试 → 不通过就重试。

默认账号密码:LangGraph 库本身无需登录。LangGraph Platform 云服务需要注册账号;本地 langgraph dev 无认证。

七、常见问题(FAQ)

Q1:LangGraph 和 LangChain 是什么关系?

LangChain 是基础组件库(LLM 封装、工具、Retriever);LangGraph 是编排层,用 LangChain 的组件搭状态图。新项目建议直接 LangGraph + langchain-core。

Q2:为什么我的 Agent 死循环了?

ReAct 循环需要一个终止条件。可以用 graph.invoke(..., config={"recursion_limit": 25}) 限制最大步数,或在条件边里判断"是否已经回答完"。

Q3:怎么部署到生产?

三种方式:(1) 直接把 graph 当 Python 函数包进 FastAPI;(2) 用官方 langgraph build 打包成 Docker;(3) 用 LangGraph Platform 云托管。

Q4:状态太大怎么办?

State 里不要塞大对象(比如整篇文档)。大内容用 Checkpointer 的 storage 单独存,State 里只放引用 ID。

Q5:和 CrewAI、AutoGen 比?

LangGraph 最底层、最可控,适合复杂生产系统;CrewAI 最业务友好,适合快速搭团队;AutoGen 最自由,适合做研究实验。

Q6:支持流式输出到前端吗?

支持。用 graph.stream(..., stream_mode="messages") 可以拿到 token-level 流式,配合 FastAPI WebSocket 或 SSE 推给前端。

Q7:LangSmith 必须用吗?

不是必须,但强烈推荐。开启后每个节点的输入输出、耗时、token 都可视化,调试效率提升 10 倍。免费额度足够个人用。

八、跳转链接

评论交流

文章目录