Hcc的Blog

嵌入式 · AI · 折腾不止

0%

LangGraph 学习(基础)

发表于 分类于 ，

简介

  LangGraph 是由 LangChain 团队开发的一个低层级 Agent 编排框架，专为构建有状态（Stateful）、长时运行的 AI 工作流而设计。与传统的线性 LLM 调用链不同，LangGraph 将工作流建模为有向图（Directed Graph）：

  • 节点（Node）：执行具体操作的函数（如调用 LLM、执行工具、处理数据）
  • 边（Edge）：定义节点之间的流转路径，支持条件分支
  • 状态（State）：在整个工作流中共享并传递的数据

  想象你正在指挥一场交响乐演出：传统的 LLM Chain 就像演奏一首从头到尾的曲子，只能顺序播放；而 LangGraph 则像一位指挥家，可以根据现场观众的反应随时调整演奏顺序，让某个乐章重复，或者跳转到特定段落。它让 AI 工作流拥有了”指挥”的智慧——能够循环、分支、回溯，真正实现复杂的自主决策。

  下面对比了 LLM Chain 与 LangGraph 的主要差异：

特性 传统 LLM Chain LangGraph
工作流结构 线性、单向执行 图结构、支持循环
状态管理 需手动实现 内置状态持久化
条件路由 实现复杂 原生支持
人机协作 需要额外开发 内置支持 interrupt
多 Agent 协调 实现困难 一流支持

  例图如下：
  

图 1 示意图
  

核心概念

  在开始编码前，我们先理解 LangGraph 的四大核心概念。

Graph (图)

  Graph 是整个工作流的蓝图，定义了 Agent 的完整逻辑结构。它由节点（Nodes）和边（Edges）组成：

1
2
3
4
5
6
7
8
9
10
StateGraph
   |-- Nodes（节点）
   |     |-- node_a
   |     |-- node_b
   |     +-- node_c
   +-- Edges（边）
         |-- START -> node_a
         |-- node_a -> node_b（条件边）
         |-- node_a -> node_c（条件边）
         +-- node_b -> END

State (状态)

  State 是贯穿整个图的共享数据结构。每个节点可以读取和更新 State，更新后的 State 会传递给下一个节点。

1
2
3
4
5
6
7
from typing import TypedDict, Annotated
from langgraph.graph import add_messages

class MyState(TypedDict):
    messages: Annotated[list, add_messages]  # 消息列表（自动追加）
    user_name: str                            # 用户名称
    step_count: int                           # 步骤计数

  Annotated[list, add_messages] 表示该字段使用 add_messages 作为 reducer ———— 新消息会追加到列表而不是覆盖。这是 LangGraph 状态管理的核心机制。

Nodes (节点)

  节点就是普通的 Python 函数，接收当前 State，返回更新后的 State（部分字段）。

1
2
3
4
5
6
7
8
9
def my_node(state: MyState) -> dict:
    # 读取状态
    messages = state["messages"]
    
    # 执行操作...
    result = "处理结果"
    
    # 返回更新的字段（不需要返回所有字段）
    return {"messages": [{"role": "ai", "content": result}]}

Edges (边)

  边定义节点之间的流转方式：

  • 普通边：固定路径，node_a -> node_b
  • 条件边：根据 State 动态路由，node_a -> node_b 或 node_c
  • 起始边：Start -> 第一个节点
  • 结束边：某节点 -> END

环境搭建

  安装 LangGraph 及其相关依赖：

1
2
3
python -m venv venv
pip install langgraph langchain langchain-openai python-dotenv -i https://mirrors.aliyun.com/pypi/simple/
pip install langgraph-cli jupyter -i https://mirrors.aliyun.com/pypi/simple/

  然后在项目根目录创建 .env 文件，写入如下内容：

1
2
3
DEEPSEEK_API_KEY=sk-xxx
DEEPSEEK_BASE_URL=https://api.deepseek.com
DEEPSEEK_MODEL=deepseek-v4-flash

  写入如下代码验证安装：

1
2
3
4
5
6
7
8
9
from importlib.metadata import version, PackageNotFoundError

try:
    v = version("langgraph")
    print(f"LangGraph 版本: {v}")
except PackageNotFoundError:
    print("错误: 未安装 langgraph 包。")
except Exception as e:
    print(f"获取版本时出错: {e}")

  我们已经成功安装了 1.2.9 版本的 LangGraph。

第一个 LangGraph 程序

  让我们从最简单的例子开始 ———— 一个只有两个节点的线性工作流。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
from langgraph.graph import StateGraph, START, END
from typing import TypedDict

# 定义 State
class SimpleState(TypedDict):
    message: str
    processed: bool

def greet_node(state: SimpleState) -> dict:
    """欢迎节点，生成问候语"""
    print(f"[greet_node] 收到消息:{state['message']}")
    return {"message":f"你好！{state['message']}"}

def process_node(state: SimpleState) -> dict:
    """处理节点，标记为已处理"""
    print(f"[process_node] 处理信息:{state['message']}")
    return {"processed": True}

builder = StateGraph(SimpleState)

# 添加节点
builder.add_node("greet", greet_node)
builder.add_node("process",process_node)

# 添加边
builder.add_edge(START, "greet")
builder.add_edge("greet", "processe")
builder.add_edge("processe", END)

# 编译成图
graph = builder.compile()

# 运行
result = graph.invoke({
    "message": "世界",
    "processed": False
})
print(f"\n最终结果: {result}")

  在学习过程中我产生了两个问题：
  1、为什么节点函数返回的是字典？和 SimpleState 是什么关系？
  这涉及到了 LangGraph 的增量更新机制。在 LangGraph 中，节点函数不需要返回完整的 State，只需要返回你想要修改的那部分字段。框架会自动将你返回的字典合并（merge） 到当前 State 中。

  2、graph.invoke() 返回的是什么类型？
  返回的是最终的完整 State，运行时类型是 dict，是符合 SimpleState 结构的字典

State 状态管理

  我们可以使用 TypeDict 定义状态：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
from typing import TypedDict, Annotated, Optional
from langgraph.graph.message import add_messages

class AgentState(TypedDict):
    # 消息历史（add_messages reducer 自动追加而非覆盖）
    messages: Annotated[list, add_messages]
    
    # 普通字段（直接覆盖）
    user_id: str
    session_id: str
    
    # 可选字段
    error: Optional[str]
    
    # 计数器（使用 operator.add 作为 reducer）
    retry_count: Annotated[int, lambda x, y: x + y]

  我们还可以使用 Pydantic 定义状态：

1
2
3
4
5
6
7
8
9
10
11
from pydantic import BaseModel, Field
from typing import Annotated
from langgraph.graph.message import add_messages

class ProductionState(BaseModel):
    messages: Annotated[list, add_messages] = Field(default_factory=list)
    user_id: str = ""
    confidence_score: float = 0.0
    
    class Config:
        arbitrary_types_allowed = True

  这两者有什么区别？Pydantic 可以真正执行验证、转换和序列化，当运行如下代码时会出现以下情况：

1
2
3
4
5
6
class UserPydantic(BaseModel):
    name: str
    age: int

u2 = UserPydantic(name="Alice", age="不是数字") # 报错
u3 = UserPydantic(name="Alice", age="25")  # 自动转换 "25" → 25

Nodes 节点

  Nodes 节点主要有以下分类

普通函数节点

  我们可以将普通函数作为节点，示例代码如下：

1
2
3
4
5
6
7
8
9
10
11
def simple_node(state: AgentState) -> dict:
    # 读取状态
    last_message = state["messages"][-1]
    
    # 执行操作
    response = f"收到: {last_message.content}"
    
    # 返回部分状态更新
    return {
        "messages": [{"role": "assistant", "content": response}]
    }

LLM 调用节点

  使用下面的模板函数可以实现 LLM 调用的节点

1
2
3
4
5
6
7
8
9
10
11
def llm_node(state: dict) -> dict:
    """调用 LLM 的节点"""
    system_prompt = SystemMessage(content="你是一个有帮助的助手。")
   
    # 将系统提示与对话历史合并
    messages = [system_prompt] + state["messages"]
   
    # 调用 LLM
    response = llm.invoke(messages)
   
    return {"messages": [response]}

异步节点

  异步节点适合 I/O 密集型操作：

1
2
3
4
5
6
7
8
9
10
11
12
import asyncio

async def async_node(state: AgentState) -> dict:
    """异步节点，适合 I/O 密集型操作"""
    # 模拟异步操作（如 API 调用、数据库查询）
    await asyncio.sleep(0.1)
    
    result = await some_async_api_call(state["messages"][-1].content)
    return {"messages": [{"role": "assistant", "content": result}]}

# 使用异步图
result = await graph.ainvoke({"messages": [...]})

使用类作为节点

  在定义了 __call__ 后，类实例可以像函数一样被调用，实际执行的是定义在类中的 __call__ 方法。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
class RouterNode:
    def __init__(self, llm, system_prompt: str):
        self.llm = llm
        self.system_prompt = system_prompt
    
    def __call__(self, state: AgentState) -> dict:
        """类实例可以作为节点使用"""
        messages = [
            SystemMessage(content=self.system_prompt),
            *state["messages"]
        ]
        response = self.llm.invoke(messages)
        return {"messages": [response]}

# 添加类节点
router = RouterNode(llm, "你是一个专业的路由助手。")
builder.add_node("router", router)

Edges 边与条件路由

普通边

1
2
3
4
5
# 固定路径：node_a 完成后始终执行 node_b
builder.add_edge("node_a", "node_b")

# 结束：node_a 完成后图结束
builder.add_edge("node_a", END)

条件边

  条件边是 LangGraph 的核心功能，根据当前 State 动态决定下一步：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
def route_after_llm(state: AgentState) -> str:
    """
    路由函数：根据 LLM 的最新输出决定走哪条路径
    返回值必须是已注册节点名称或 END
    """
    last_message = state["messages"][-1]
    
    # 如果 LLM 请求使用工具
    if hasattr(last_message, "tool_calls") and last_message.tool_calls:
        return "tools"
    
    # 否则结束
    return END

# 添加条件边
builder.add_conditional_edges(
    "llm",              # 源节点
    route_after_llm,    # 路由函数
    {
        "tools": "tool_executor",   # 返回 "tools" 时 -> tool_executor 节点
        END: END                    # 返回 END 时 -> 结束
    }
)

Human-in-the-Loop 人机协作

  LangGraph 原生支持在工作流执行过程中暂停，等待人工审核或输入。这对于需要人工确认的敏感操作非常有用。
  我们可以使用 interrupt 实现暂停执行：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
from langgraph.types import interrupt
from langgraph.checkpoint.memory import MemorySaver

def sensitive_action_node(state: MessagesState) -> dict:
    """执行敏感操作前请求人工审批"""
    last_msg = state["messages"][-1].content
    
    # 暂停图的执行，等待人工决策
    human_decision = interrupt({
        "question": "是否批准执行以下操作？",
        "action": last_msg,
        "risk_level": "中等"
    })
    
    if human_decision == "approve":
        return {"messages": [{"role": "assistant", "content": "操作已批准并执行完毕。"}]}
    else:
        return {"messages": [{"role": "assistant", "content": "操作已取消。"}]}

# 必须使用 checkpointer 才能支持 interrupt
checkpointer = MemorySaver()
graph = builder.compile(checkpointer=checkpointer)

  另一种方式，我们可以在边上也设置断点：

1
2
3
4
5
6
# 另一种方式：在编译时指定断点
graph = builder.compile(
    checkpointer=checkpointer,
    interrupt_before=["sensitive_node"],   # 执行该节点前暂停
    # interrupt_after=["review_node"],     # 执行该节点后暂停
)

  **checkpointer 是实现 interrupt 功能的前提条件。**没有 checkpointer，图无法保存暂停时的状态，也就无法在恢复时继续执行。MemorySaver 适合开发测试，生产环境建议使用 SqliteSaver 或其他持久化存储。

多 Agent 系统

  LangGraph 擅长协调多个专门化的 Agent 协同工作。通过将复杂任务分解给不同的专家 Agent，可以实现更强大的问题解决能力。

主从架构

  我们常常使用的架构是主从架构，有一个主管 Agent 负责拆解任务并分配给各个子 Agent。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
import os
from dotenv import load_dotenv
from langgraph.graph import StateGraph, MessagesState, START, END
from langchain_openai import ChatOpenAI
from langchain_core.messages import SystemMessage, HumanMessage

load_dotenv()

# 初始化 LLM
llm = ChatOpenAI(
    model=os.getenv('DEEPSEEK_MODEL', 'deepseek-v4-pro'),
    openai_api_key=os.getenv('DEEPSEEK_API_KEY'),
    openai_api_base=os.getenv('DEEPSEEK_BASE_URL', 'https://api.deepseek.com'),
    temperature=0
)

# 定义专家 Agent
def research_agent(state: MessagesState) -> dict:
    """研究 Agent：负责信息收集"""
    system = SystemMessage(content="你是一个专业的研究员，负责收集和整理信息。请简洁地总结关键信息。")
    response = llm.invoke([system] + state["messages"])
    return {"messages": [response]}

def writing_agent(state: MessagesState) -> dict:
    """写作 Agent：负责内容创作"""
    system = SystemMessage(content="你是一个专业的写作者，负责根据已有信息撰写内容。请保持内容清晰流畅。")
    response = llm.invoke([system] + state["messages"])
    return {"messages": [response]}

def review_agent(state: MessagesState) -> dict:
    """审校 Agent：负责质量控制"""
    system = SystemMessage(content="你是一个专业的编辑，负责审核和改进内容质量。请指出问题并给出改进建议。")
    response = llm.invoke([system] + state["messages"])
    return {"messages": [response]}

# 主管 Agent 决定流程
def supervisor_node(state: MessagesState) -> dict:
    """主管：协调各专家 Agent 的工作"""
    system = SystemMessage(content="""你是一个工作流主管。
根据任务进度决定下一步应该由哪个 Agent 处理。
分析对话历史，只返回以下之一：RESEARCH、WRITING、REVIEW、FINISH
- RESEARCH：需要收集更多信息
- WRITING：信息充足，可以开始写作
- REVIEW：写作完成，需要审核
- FINISH：任务已完成
""")
    response = llm.invoke([system] + state["messages"])
    return {"messages": [response]}

def route_by_supervisor(state: MessagesState) -> str:
    """根据主管决策路由"""
    last_msg = state["messages"][-1].content.strip().upper()
    
    if "RESEARCH" in last_msg:
        return "research"
    elif "WRITING" in last_msg:
        return "writing"
    elif "REVIEW" in last_msg:
        return "review"
    else:
        return END

# 构建多 Agent 图
builder = StateGraph(MessagesState)
builder.add_node("supervisor", supervisor_node)
builder.add_node("research", research_agent)
builder.add_node("writing", writing_agent)
builder.add_node("review", review_agent)

builder.add_edge(START, "supervisor")
builder.add_conditional_edges("supervisor", route_by_supervisor)

# 每个专家完成后返回主管
for agent in ["research", "writing", "review"]:
    builder.add_edge(agent, "supervisor")

graph = builder.compile()

# 测试多 Agent 协作
result = graph.invoke({
    "messages": [HumanMessage(content="请帮我写一篇关于 Python 装饰器的简短介绍文章")]
})

print("=== 多 Agent 协作完成 ===")
for i, msg in enumerate(result["messages"]):
    print(f"\n[{i+1}] {msg.type}: {msg.content[:150]}...")

子图

  我们还可以将复杂的子流程封装为子图，并在主图中复用。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
# 将复杂子流程封装为子图，在主图中复用
sub_builder = StateGraph(MessagesState)
sub_builder.add_node("step1", step1_node)
sub_builder.add_node("step2", step2_node)
sub_builder.add_edge(START, "step1")
sub_builder.add_edge("step1", "step2")
sub_builder.add_edge("step2", END)
sub_graph = sub_builder.compile()

# 在主图中使用子图
main_builder = StateGraph(MessagesState)
main_builder.add_node("preprocessing", preprocess_node)
main_builder.add_node("sub_workflow", sub_graph)  # 直接使用编译好的子图
main_builder.add_node("postprocessing", postprocess_node)

main_builder.add_edge(START, "preprocessing")
main_builder.add_edge("preprocessing", "sub_workflow")
main_builder.add_edge("sub_workflow", "postprocessing")
main_builder.add_edge("postprocessing", END)

main_graph = main_builder.compile()

总结

  本次我们学习了 LangGraph 的基本用法，在日后的学习中会使用 LangChain 做一些基本的 Agent 开发。