结合实操逻辑与代码示例，分步讲解 LangGraph 定义图、分支、循环、人在回路、执行流程，基于官方标准写法，通俗易懂 + 可落地。

一、前置基础

核心四要素：

1. State：全局状态（所有节点读写的统一数据容器）
2. Node：执行单元（函数 / 逻辑块）
3. Edge：流向（普通边、条件边）
4. Graph：状态图本体，串联所有节点与流向

依赖包：langgraph、langchain、langchain-openai

---

二、1. 定义状态（State）

所有节点共享同一份状态，是数据流转的载体，使用 TypedDict 定义。

python

运行

from typing import TypedDict, Annotated, Sequence
import operator

# 定义全局状态
class GraphState(TypedDict):
    messages: Annotated[Sequence[str], operator.add]  # 对话消息，自动追加
    query: str                                       # 用户问题
    tool_result: str                                 # 工具返回结果
    final_answer: str                                # 最终答案

• Annotated[..., operator.add]：列表类型字段，新内容追加而非覆盖，适配多轮对话。

---

三、2. 定义节点（Node）

节点是独立执行逻辑，入参 = State，出参 = 更新后的 State 字典。 示例创建 3 个基础节点：

python

运行

# 节点1：大模型思考判断
def llm_think(state: GraphState) -> dict:
    print("【LLM 思考节点】")
    return {"messages": ["LLM 完成思考"]}

# 节点2：调用工具（检索/查询）
def call_tool(state: GraphState) -> dict:
    print("【工具调用节点】")
    return {"tool_result": "工具查询到的内容"}

# 节点3：生成最终答案
def generate_answer(state: GraphState) -> dict:
    print("【答案生成节点】")
    return {"final_answer": "基于信息生成最终回答"}

---

四、3. 定义图（Graph）本体

使用 StateGraph 基于自定义 State 创建图结构，步骤：实例化 → 添加节点 → 设置入口。

python

运行

from langgraph.graph import StateGraph, START, END

# 1. 实例化状态图，绑定自定义State
workflow = StateGraph(GraphState)

# 2. 把上面定义的节点注册到图中（名称 + 函数）
workflow.add_node("llm_think", llm_think)
workflow.add_node("call_tool", call_tool)
workflow.add_node("generate_answer", generate_answer)

# 3. 设置图的入口节点（流程起点）
workflow.add_edge(START, "llm_think")

• START：图内置起点；END：图内置终点。

---

五、4. 定义分支（条件路由）

分支依靠 条件边（Conditional Edge） 实现：根据 State 中的字段，动态选择下一个节点。

4.1 编写分支判断函数

函数入参为 State，返回目标节点名称字符串。

python

运行

def route_by_content(state: GraphState) -> str:
    """分支判断逻辑：模拟判断是否需要调用工具"""
    # 业务规则：有查询问题 → 走工具；无 → 直接生成答案
    if state.get("query"):
        return "call_tool"
    else:
        return "generate_answer"

4.2 绑定条件分支到图

python

运行

# 从 llm_think 节点出发，执行分支判断
workflow.add_conditional_edges(
    source="llm_think",        # 起始节点
    path=route_by_content,     # 分支判断函数
    path_map={                 # 映射：函数返回值 → 对应节点名
        "call_tool": "call_tool",
        "generate_answer": "generate_answer"
    }
)

# 普通边：工具执行完成后，走到答案生成节点
workflow.add_edge("call_tool", "generate_answer")

# 普通边：答案生成完成 → 流程结束
workflow.add_edge("generate_answer", END)

流程逻辑： START → llm_think → 【分支判断】→ call_tool / generate_answer → END

---

六、5. 定义循环（有环流程）

循环本质：条件边 + 回指边，执行完节点后，重新回到上游节点，形成迭代。

改造：增加循环逻辑（模拟「信息不足→重新查询」）

步骤 1：新增循环判断函数

python

运行

def need_loop(state: GraphState) -> str:
    """循环判断：模拟信息是否充足"""
    # 模拟规则：工具结果为空 → 继续循环；不为空 → 结束循环
    if not state.get("tool_result"):
        return "llm_think"  # 回到思考节点，开启新一轮循环
    else:
        return "generate_answer"

步骤 2：修改流向，构建环

python

运行

# 清空原有流向，重新配置（演示循环）
workflow = StateGraph(GraphState)
workflow.add_node("llm_think", llm_think)
workflow.add_node("call_tool", call_tool)
workflow.add_node("generate_answer", generate_answer)
workflow.add_edge(START, "llm_think")

# 1. 思考 → 固定走到工具节点
workflow.add_edge("llm_think", "call_tool")

# 2. 工具执行后，进入循环判断（核心：环的入口）
workflow.add_conditional_edges(
    source="call_tool",
    path=need_loop,
    path_map={
        "llm_think": "llm_think",    # 回跳上游节点 = 循环
        "generate_answer": "generate_answer"
    }
)

# 3. 最终节点走向结束
workflow.add_edge("generate_answer", END)

循环流转图 START → llm_think → call_tool → 条件判断

• 信息不足 → 回到 llm_think（循环执行）
• 信息充足 → generate_answer → END

---

七、6. 定义人在回路（Human-in-the-Loop）

核心能力：流程暂停、人工介入、恢复执行，两种主流实现方式：

方式 1：全局暂停（推荐，简单易用）

编译图时开启 interrupt_before，指定在某个节点执行前暂停，等待人工操作。

python

运行

# 编译图，并设置：执行 call_tool 节点之前暂停
app = workflow.compile(interrupt_before=["call_tool"])

运行 + 人工交互流程

1. 首次执行，流程走到暂停点，返回当前 State；
2. 人工修改状态（补充信息、修正参数）；
3. 调用 app.invoke(None) 恢复流程继续执行。

完整交互示例

python

运行

# 初始输入
inputs = {"query": "查询数据", "tool_result": ""}

# 第一轮执行：触发暂停
config = {"configurable": {"thread_id": "1"}}  # thread_id 区分会话
state = app.invoke(inputs, config=config)
print("流程已暂停，等待人工操作")

# ========== 人工介入：修改状态 ==========
current_state = app.get_state(config)
# 人工补充工具结果
app.update_state(config, {"tool_result": "人工补充的查询内容"})

# ========== 恢复流程继续执行 ==========
final_state = app.invoke(None, config=config)
print("流程执行完毕：", final_state["final_answer"])

方式 2：自定义人工节点

新增专属人工节点，把人工操作作为普通流程节点，灵活性更高，适合复杂审核场景。

---

八、7. 执行图（完整运行方式）

图必须先 compile 编译，再执行，提供三种执行模式：

8.1 基础编译

python

运行

# 普通编译（无暂停）
app = workflow.compile()

# 带暂停（人在回路专用）
app = workflow.compile(interrupt_before=["节点名"])

8.2 三种执行 API

1）invoke 同步执行（全流程跑完，返回最终状态）

python

运行

inputs = {"query": "测试问题"}
result = app.invoke(inputs)
print(result)

2）stream 流式执行（逐节点返回结果，看中间过程）

适合调试、前端实时展示：

python

运行

for event in app.stream(inputs):
    print(event)

3）多会话隔离（thread_id）

通过 thread_id 区分不同用户 / 会话，搭配检查点可持久化状态：

python

运行

config = {"configurable": {"thread_id": "session_001"}}
app.invoke(inputs, config=config)

---

九、核心总结（速记）

1. 定义图

   1. 用 TypedDict 定义 State
   2. 编写业务 Node 函数
   3. StateGraph 实例化 + add_node 注册节点 + 设置 START 入口

2. 定义分支 编写条件判断函数 → add_conditional_edges 绑定路由，根据状态动态跳转节点。

3. 定义循环 基于条件边，让流程回跳到上游节点，形成有环图，实现迭代执行。

4. 定义人在回路 编译时配置 interrupt_before 指定暂停节点 → 执行到节点前自动暂停 → update_state 人工改状态 → 再次 invoke 恢复。

5. 执行图 先 compile 编译 → 使用 invoke/stream 运行；thread_id 实现多会话隔离。