LangGraph 基础

LangGraph 的核心是把 AI 应用拆成状态、节点和边，让复杂流程不再只是一串函数调用，而是一张可以控制、观察和扩展的工作流图。

LangGraph 是什么

LangGraph 是 LangChain 生态里的工作流编排框架。

Graph，中文就是“图”。

这里的图不是图片，而是这种结构：

节点 A
  -> 节点 B
  -> 节点 C

或者：

节点 A
  -> 判断
      -> 成功：结束
      -> 失败：重新生成

在 AI Agent 项目里，LangGraph 适合处理：

多步骤 AI 流程
条件分支
循环重试
工具调用
RAG 检索流程
人工确认
状态持久化

本节先学最小概念：

State
Node
Edge
最小工作流

为什么需要 LangGraph

普通 Chain 通常是固定顺序：

Prompt
  -> Model
  -> Parser

但真实 Agent 流程经常不是一条直线。

比如文档问答：

用户提问
  -> 改写问题
  -> 检索文档
  -> 判断文档是否足够
      -> 足够：生成回答
      -> 不足：扩大检索
  -> 返回答案

再比如售后 Agent：

用户输入
  -> 判断意图
      -> 查订单
      -> 查库存
      -> 创建工单
      -> 普通聊天
  -> 生成回复

这种流程用一堆 if else 也能写。

但当节点越来越多时，代码会变得难维护。

LangGraph 的作用就是把流程显式画出来。

安装依赖

python -m pip install langgraph langchain langchain-openai python-dotenv

如果只学习纯工作流，不调用模型，也可以只安装：

python -m pip install langgraph

State

State，中文可以理解为“状态”。

它是整条工作流共享的数据。

比如：

用户问题
初稿
最终回答
错误信息
检索结果
执行次数

用 Python 可以先用 TypedDict 定义：

from typing import TypedDict


class AgentState(TypedDict):
    question: str
    draft: str
    answer: str

这个 State 表示工作流里会有三个字段：

question：用户问题
draft：中间草稿
answer：最终答案

LangGraph 里的节点会读取 State，也会返回 State 的一部分更新。

Node

Node，中文就是“节点”。

节点本质上是一个 Python 函数。

它接收 State，返回要更新的字段。

示例：

def generate_draft(state: AgentState):
    question = state["question"]
    return {
        "draft": f"先给出一个草稿回答：{question}"
    }

再写一个润色节点：

def polish_answer(state: AgentState):
    draft = state["draft"]
    return {
        "answer": f"{draft}\n\n整理后：这是一个更清楚的回答。"
    }

注意：

节点不需要返回完整 State
只返回自己负责更新的字段

这样每个节点职责更清楚。

Edge

Edge，中文就是“边”。

它表示节点之间的执行顺序。

比如：

START
  -> generate_draft
  -> polish_answer
  -> END

START 和 END 是 LangGraph 提供的特殊节点。

含义是：

START：流程开始
END：流程结束

image.png
add_node增加node
add_edge设定顺序

最小工作流

完整代码：

from typing import TypedDict

from langgraph.graph import END, START, StateGraph


class AgentState(TypedDict):
    question: str
    draft: str
    answer: str


def generate_draft(state: AgentState):
    question = state["question"]
    return {
        "draft": f"先给出一个草稿回答：{question}"
    }


def polish_answer(state: AgentState):
    draft = state["draft"]
    return {
        "answer": f"{draft}\n\n整理后：这是一个更清楚的回答。"
    }


builder = StateGraph(AgentState)

builder.add_node("generate_draft", generate_draft)
builder.add_node("polish_answer", polish_answer)

builder.add_edge(START, "generate_draft")
builder.add_edge("generate_draft", "polish_answer")
builder.add_edge("polish_answer", END)

graph = builder.compile()

result = graph.invoke({
    "question": "什么是 LangGraph？"
})

print(result["answer"])

执行流程：

输入 question
  -> generate_draft 写入 draft
  -> polish_answer 读取 draft，写入 answer
  -> 返回最终 State

加入模型调用

节点里可以调用模型。

先创建模型：

import os

from dotenv import load_dotenv
from langchain_openai import ChatOpenAI

load_dotenv()

model = ChatOpenAI(
    model=os.getenv("QWEN_MODEL", "qwen-plus"),
    api_key=os.environ["DASHSCOPE_API_KEY"],
    base_url=os.getenv(
        "DASHSCOPE_BASE_URL",
        "https://dashscope.aliyuncs.com/compatible-mode/v1",
    ),
    temperature=0.3,
)

把节点改成真实模型调用：

def generate_draft(state: AgentState):
    response = model.invoke([
        {
            "role": "system",
            "content": "你是一个 AI 学习助手，先生成简短草稿。",
        },
        {
            "role": "user",
            "content": state["question"],
        },
    ])

    return {
        "draft": response.content
    }

润色节点：

def polish_answer(state: AgentState):
    response = model.invoke([
        {
            "role": "system",
            "content": "请把草稿整理成适合初学者阅读的回答。",
        },
        {
            "role": "user",
            "content": state["draft"],
        },
    ])

    return {
        "answer": response.content
    }

这就是一个两步 LLM 工作流。

条件边

真实工作流经常需要判断。

比如：

回答太短
  -> 重新生成
回答够长
  -> 结束

先扩展 State：

class AgentState(TypedDict):
    question: str
    draft: str
    answer: str
    retry_count: int

写一个判断函数：

def route_answer(state: AgentState):
    if len(state["answer"]) < 80 and state["retry_count"] < 1:
        return "retry"

    return "finish"

条件边：

builder.add_conditional_edges(
    "polish_answer",
    route_answer,
    {
        "retry": "generate_draft",
        "finish": END,
    },
)

这表示：

polish_answer 执行完
  -> route_answer 判断
      -> retry：回到 generate_draft
      -> finish：结束

这里要小心循环。

一定要有结束条件。

State 不是数据库

State 是一次图执行过程里的状态。

它不是业务数据库。

你可以把它理解为：

一次请求的上下文对象

它适合放：

用户问题
中间结果
检索结果
模型回答
执行次数
错误信息

不适合直接放：

大量原始文档
完整数据库结果集
大文件二进制内容
敏感凭证

如果后面需要保存长期状态，要用数据库或 LangGraph 的持久化能力。

本节先不展开。

和 Java 后端的关系

在这个项目路线里，可以这样分工：

Java Spring Boot：对外接口、权限、用户、数据库、文件上传
Python FastAPI：封装 AI 工作流
LangGraph：编排 Python 里的 Agent/RAG 多步骤流程

调用链路：

前端
  -> Java Spring Boot
  -> Python FastAPI
  -> LangGraph workflow
  -> 模型 / 工具 / 检索

LangGraph 不一定直接暴露给前端。

它通常藏在 Python 服务内部。

一个完整 demo

新建 langgraph_basic_demo.py：

from typing import TypedDict

from langgraph.graph import END, START, StateGraph


class AgentState(TypedDict):
    question: str
    draft: str
    answer: str


def generate_draft(state: AgentState):
    return {
        "draft": f"草稿：{state['question']} 可以理解为一个由状态驱动的工作流。"
    }


def polish_answer(state: AgentState):
    return {
        "answer": f"{state['draft']}\n\n最终回答：LangGraph 用节点和边组织 AI 应用流程。"
    }


builder = StateGraph(AgentState)
builder.add_node("generate_draft", generate_draft)
builder.add_node("polish_answer", polish_answer)
builder.add_edge(START, "generate_draft")
builder.add_edge("generate_draft", "polish_answer")
builder.add_edge("polish_answer", END)

graph = builder.compile()

result = graph.invoke({
    "question": "什么是 LangGraph？"
})

print(result)

运行：

python langgraph_basic_demo.py

常见问题

LangGraph 和 LangChain 是竞争关系吗

不是。

LangChain 更偏应用组件组合。

LangGraph 更偏工作流控制。

很多 LangChain Agent 底层也会使用 LangGraph 的能力。

什么场景先不用 LangGraph

如果只有：

一个 prompt
一次模型调用
一个简单工具

先不用 LangGraph。

等流程出现分支、循环、多节点协作时再引入。

节点里能不能调用数据库

可以。

但要注意：

节点职责要小
异常要处理
外部调用要加超时
结果不要无限塞进 State

条件边是不是就是 if else

本质上是。

区别是 LangGraph 把 if else 变成了图的一部分。

这样流程更容易观察、测试和扩展。

练习清单

完成几件事情：

安装 langgraph
定义一个 TypedDict State
写两个 Node 函数
用 StateGraph 注册节点
用 START 和 END 连接流程
compile 生成 graph
用 invoke 执行一次工作流
给节点接入 ChatOpenAI
写一个条件边
能解释 State、Node、Edge 的区别

建议目录：

ai-agent-study
├── python
│   └── langgraph-basic
│       ├── langgraph_basic_demo.py
│       ├── requirements.txt
│       └── README.md
└── docs
    └── langgraph-basic.md

小结

本节的结论：

LangGraph 用 State 保存上下文，用 Node 执行步骤，用 Edge 控制流程。

最小链路：

START
  -> Node A
  -> Node B
  -> END

带条件链路：

Node
  -> route
      -> 分支 A
      -> 分支 B

image.png
后面做 RAG 和 Agent 时，LangGraph 会用来组织更复杂的多步骤流程。

参考资料

这个家伙很懒，啥也没有留下😋