Jiaming Zhang's Blog
0%

LangGraph 初体验

发表于 更新于

初步学习体验 LangGraph

先了解下 LangGraph 在 AI 智能体生态中的定位

构建大模型应用/AI智能体主要由如下几种方式：

类别 特点 代表产品 目标群体
低代码可视化平台 通过 UI 拖拽编排工作流 Dify, Coze, n8n 开发者，产品经理，企业团队
通用的 Agent 软件 直接使用通用能力 OpenClaw, Claude Code -
开发框架 手撕代码，定制化开发 LangChain, LangGraph, DeepAgents, Spring AI AI 应用开发者

之前已介绍过 LangChain，本文只介绍 LangGraph

产生背景

已经有 langchain，为什么要有 langGraph？

先看下 LangChain 解决了哪些问题：

  1. 封装主流大模型，方便调用
  2. Prompt 工程化
  3. 工具调用
  4. 上下文管理
  5. 封装 RAG（检索增强生成）

LangChain 让调用大模型像搭积木一样，将不同组件连起来，随着功能的复杂，Langchain 出现了一些问题：

  1. 无法处理复杂流程：LangChain 适合线性流程，对于分支、循环、状态判断，开发者只好自己处理
  2. 流程不可控：下一步使用哪个工具，完全交给 LLM，LLM 幻觉很可能导致死循环，执行路径不可复现
  3. 任务无法中断：无法保存/恢复任务的中间状态
  4. 无法观测执行路径，调试困难

比如用户想购买某款产品，需要我们咨询用户的需求和偏好，推荐几款热门产品

这显示无法用线性方案实现，因为用户不会按照预想的流程配合我们

而是一个循环引导的过程

再比如我们有很多工具，有些工具之间存在固定的依赖关系，如果把所有工具一股脑丢给 llm，完全由 llm 决定如何组织，得到的结果非常不确定；如果把工具分类成组，每个组由一个 llm 管理，形成若干 子Agent，再将子Agent 连起来，会让工具执行更准确可控

LangGraph 正是为解决上述问题

核心概念

图（Graph）

LangGraph 将程序定义为图，图由节点（Nodes）和边（Edges）组成

  • 节点（Nodes）：代表一个执行步骤
  • 边（Edges）：定义节点之间的转换路径，边可以是有条件的

状态（State）

状态是图在执行过程中的共享数据结果（Context 上下文）

  • 每个节点执行完毕，都会更新状态
  • 所有节点都能访问最新状态

消息 (Message)

消息用来封装不同角色的输入与输出，封装图中传输的数据，分为

  • SystemMessage 系统设定 Prompt
  • HumanMessage 用户输入的消息
  • AIMessage 模型回复的消息
  • ToolMessage 工具调用返回的消息

下面写一个 demo 体验下 langGraph 如何构建 workflow 与 Agent

Workflow 与 Agent 的区别：

  • Agent：由 llm 决定如何使用工具来完成任务
  • workflow：一系列明确的步骤自动化执行工具，Agent 可以作为 workflow 中的一个组件

Workflow Demo

场景：安排行程的小助理，根据我的要求，推荐出行方案，根据我的选择下单

流程：用户输入，如果用户输入存在时间、起点、终点，返回推荐方案，如果缺少某个信息，引导用户输入，直到输入完整再返回推荐

项目初始化

https://docs.langchain.com/oss/python/langgraph/local-server

代码

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
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
from langchain.chat_models import init_chat_model
from typing_extensions import TypedDict
from typing import Optional, List
from langgraph.graph import StateGraph, START, END
from langgraph.types import interrupt
import json


llm = init_chat_model(
        "deepseek-chat", 
        model_provider="deepseek", 
        streaming=True, 
        temperature=0.7
    )

class TravelState(TypedDict):
    messages: str
    origin: Optional[str]
    destination: Optional[str]
    time: Optional[str]
    choice: Optional[int]
    plans: Optional[List[str]]
    choice: Optional[int]
    selected_plan: Optional[str]
    order: Optional[str]
    missing: Optional[List[str]]
    
def parse_query(state: TravelState):
    text = state["messages"]
    print(text)
    
    prompt = f"""
      你是一个信息抽取系统。

      任务：从用户输入中提取出行信息：

      字段定义：
      - origin：出发地（城市名）
      - destination：目的地（城市名）
      - time：时间（如：今天、明天、后天、具体日期）

      规则：
      1. 只要用户“可能表达了”，就要提取（不要过于保守）
      2. 可以做合理推断（如“去上海” → destination=上海）
      3. 不要全部返回 null，除非真的完全没有信息
      4. 只返回 JSON，不要解释

      用户输入：
      {text}

      输出格式：
      {{
        "origin": "...",
        "destination": "...",
        "time": "..."
      }}
      """
    
    resp = llm.invoke(prompt)
    print(resp)
    
    try:
        data = json.loads(resp.content)
    except:
        data = {}
    
    # 和旧 state 合并
    return {
        "origin": data.get("origin") or state.get("origin"),
        "destination": data.get("destination") or state.get("destination"),
        "time": data.get("time") or state.get("time"),
    }

def check_missing(state: TravelState):
    missing = []
    if not state.get('origin'):
        missing.append('出发地')
    if not state.get('destination'):
        missing.append('目的地')
    if not state.get('time'):
        missing.append('时间')
    return {'missing': missing}

def ask_user(state: TravelState):
    missing = state['missing']
    question = f'请补充以下信息:{', '.join(missing)}'
    return interrupt(question)

def route_after_check(state: TravelState):
    if state['missing']:
        return 'ask'
    return 'generate'

def generate_plans(state: TravelState):
    plans = [
        f"{state['origin']}→{state['destination']} A101 3h 500元",
        f"{state['origin']}→{state['destination']} B293 2h30min 600元",
        f"{state['origin']}→{state['destination']} C987 3h20min 559元",
    ]
    return {'plans': plans}
  
# 构建图
builder = StateGraph(TravelState)

builder.add_node("parse", parse_query)
builder.add_node("check", check_missing)
builder.add_node("ask", ask_user)
builder.add_node('generate', generate_plans)

builder.set_entry_point('parse')
builder.add_edge('parse', 'check')
builder.add_conditional_edges(
    'check',
    route_after_check,
    {
        'ask': 'ask',
        "generate": 'generate'
    }
)

builder.add_edge('ask', 'parse')
# 部分代码逻辑重复未贴出

agent = builder.compile()

from langchain_core.messages import HumanMessage
messages = [HumanMessage(content="请帮我安排行程")]
messages = agent.invoke({"messages": messages})

上面代码中，llm 用于解析用户输入的自然语言，提取关键数据，其他 node 用关键数据做业务逻辑，edge 负责控制 node 的执行

LangSmith 调试

LangSmith 是由 LangChain 推出的用于构建、调试、评估和监控大语言模型的平台

调试：可以记录每一步做了什么，输入输出

追踪：可以将复杂调用流程可视化

评估：对比不同 prompt、模型，自动打分

监控：统计调用成功率，延迟，token使用量，异常等

调试过程：

1
2
# 输入命令，会自动打开 LangSmith 网页
langgraph dev

调试过程：

输入 调试
{"messages": "请帮我安排明天的行程"}
{"messages": "从上海到北京"}

上面的流程显得比较麻烦，因为每个表单填充都需要实现 input -> parse -> check -> ask -> parse ... 这个固定过程，可以用 Slot Filling（槽位填充）简单实现，这里只是为了体现 langGraph 的 node 和 edge

Agent demo

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
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
# Step 1: Define tools and model

from langchain.tools import tool
from langchain.chat_models import init_chat_model
from langchain_core.messages import HumanMessage, ToolMessage, AnyMessage, SystemMessage
from typing_extensions import TypedDict, Annotated
import operator
from typing import Literal
from langgraph.graph import StateGraph, START, END

model = init_chat_model(
        "deepseek-chat", 
        model_provider="deepseek", 
        streaming=True, 
        temperature=0.7
    )


# Define tools
@tool
def multiply(a: int, b: int) -> int:
    """Multiply `a` and `b`.

    Args:
        a: First int
        b: Second int
    """
    return a * b


@tool
def add(a: int, b: int) -> int:
    """Adds `a` and `b`.

    Args:
        a: First int
        b: Second int
    """
    return a + b


@tool
def divide(a: int, b: int) -> float:
    """Divide `a` and `b`.

    Args:
        a: First int
        b: Second int
    """
    return a / b


# Augment the LLM with tools
tools = [add, multiply, divide]
tools_by_name = {tool.name: tool for tool in tools}
model_with_tools = model.bind_tools(tools)

# Step 2: Define state

class MessagesState(TypedDict):
    messages: Annotated[list[AnyMessage], operator.add]
    llm_calls: int

# Step 3: Define model node


def llm_call(state: dict):
    """LLM decides whether to call a tool or not"""

    return {
        "messages": [
            model_with_tools.invoke(
                [
                    SystemMessage(
                        content="You are a helpful assistant tasked with performing arithmetic on a set of inputs."
                    )
                ]
                + state["messages"]
            )
        ],
        "llm_calls": state.get('llm_calls', 0) + 1
    }


# Step 4: Define tool node

def tool_node(state: dict):
    """Performs the tool call"""

    result = []
    for tool_call in state["messages"][-1].tool_calls:
        tool = tools_by_name[tool_call["name"]]
        observation = tool.invoke(tool_call["args"])
        result.append(ToolMessage(content=observation, tool_call_id=tool_call["id"]))
    return {"messages": result}

# Step 5: Define logic to determine whether to end


# Conditional edge function to route to the tool node or end based upon whether the LLM made a tool call
def should_continue(state: MessagesState) -> Literal["tool_node", END]:
    """Decide if we should continue the loop or stop based upon whether the LLM made a tool call"""

    messages = state["messages"]
    last_message = messages[-1]

    # If the LLM makes a tool call, then perform an action
    if last_message.tool_calls:
        return "tool_node"

    # Otherwise, we stop (reply to the user)
    return END

# Step 6: Build agent

# Build workflow
agent_builder = StateGraph(MessagesState)

# Add nodes
agent_builder.add_node("model", llm_call)
agent_builder.add_node("tools", tool_node)

# Add edges to connect nodes
agent_builder.add_edge(START, "tools")
agent_builder.add_conditional_edges(
    "llm_call",
    should_continue,
    ["tool_node", END]
)
agent_builder.add_edge("tools", "model")

# Compile the agent
agent = agent_builder.compile()


from IPython.display import Image, display
# Show the agent
display(Image(agent.get_graph(xray=True).draw_mermaid_png()))

# Invoke
messages = [HumanMessage(content="Add 3 and 4.")]
messages = agent.invoke({"messages": messages})
for m in messages["messages"]:
    m.pretty_print()

调试效果

执行流程 耗时与token统计

上面的代码手动实现了 Agent ReAct 模式，只为说明 langGraph 的运行模式，langChain 已经为我们封装好了创建 Agent 方法

1
2
3
4
5
6
from langchain.agents import create_agent
agent = create_agent(
    model=model,
    tools=tools,
    system_prompt="You are a helpful assistant."
)

小结

最后总结下 LangGraph 的

优点：

  1. 流程可控，支持复杂流程
  2. 显式状态管理：可中断，可持久化
  3. 易调试

使用场景：

  1. 复杂的多分支流程
  2. 多 Agent 协作
  3. 长作业，中间需要人工介入