第 10 章 复盘：LangGraph 核心概念 & 第一个 Graph#

1
import asyncio
2

3
from langgraph.graph import StateGraph, END
4

5

6
async def build_and_run_simple_graph():
7
    # 第一步：创建 StateGraph，传入 State 类型
8
    builder = StateGraph(ChatState)
9

10
    # 第二步：添加节点 add_node(节点名称, 节点函数)
11
    builder.add_node("preprocess", preprocess_node)
12
    builder.add_node("llm", llm_node)
13

14
    # 第三步：设置入口节点
15
    builder.set_entry_point("preprocess")
16

17
    # 第四步：添加边（定义执行顺序）
18
    builder.add_edge("preprocess", "llm")   # preprocess 执行完 → llm
19
    builder.add_edge("llm", END)            # llm 执行完 → 结束（END 为内置终止节点）
20

21
    # 第五步：编译，得到可执行的 CompiledGraph
22
    graph = builder.compile()
23

24
    # 第六步：打印 Mermaid 图，可视化结构（可粘贴到 https://mermaid.live）
25
    print("Graph 结构（Mermaid 格式）：")
26
    print(graph.get_graph().draw_mermaid())
27

28
    # 第七步：ainvoke 传入初始 State 运行
29
    initial_state = {
30
        "messages": [HumanMessage(content="你好，用一句话介绍一下 LangGraph")],
31
        "current_step": "start",
32
    }
33
    print("\n开始运行 Graph...")
34
    result = await graph.ainvoke(initial_state)
35

36
    print(f"\n最终 State:")
37
    print(f"  current_step: {result['current_step']}")
38
    print(f"  messages 数量: {len(result['messages'])}")
39
    print(f"  最后一条消息: {result['messages'][-1].content}")
40

41

42
asyncio.run(build_and_run_simple_graph())

1
uv run python scripts/ch10/test_basic_graph.py

1
import asyncio
2
from typing import TypedDict, Annotated, Literal
3
import operator
4

5
from langchain_core.messages import BaseMessage, HumanMessage
6
from langgraph.graph import StateGraph, END
7

8
from app.ai.llm import llm
9

10

11
class AgentState(TypedDict):
12
    # 消息列表：追加模式
13
    messages: Annotated[list[BaseMessage], operator.add]
14
    # 下一步动作标记：need_tool / need_refine / direct
15
    next_action: str
16
    # 工具调用次数：防止工具环节无限循环
17
    tool_call_count: int
18

19

20
async def analyze_node(state: AgentState) -> dict:
21
    """分析节点：判断该走哪条路径"""
22
    print("[分析节点] 判断问题类型...")
23
    question = state["messages"][-1].content
24
    # 这里用规则模拟路由判断；实际项目中可由 LLM 输出动作标记
25
    if "天气" in question:
26
        return {"next_action": "need_tool"}
27
    return {"next_action": "direct"}
28

29

30
async def direct_answer_node(state: AgentState) -> dict:
31
    """直接回答节点"""
32
    print("[直接回答节点] 生成回复...")
33
    response = await llm.ainvoke(state["messages"])
34
    return {"messages": [response], "next_action": "done"}
35

36

37
async def tool_node(state: AgentState) -> dict:
38
    """工具节点：模拟调用外部工具"""
39
    print("[工具节点] 调用工具获取数据...")
40
    tool_result = HumanMessage(content="（工具返回）武汉今天晴，25~32℃")
41
    return {
42
        "messages": [tool_result],
43
        "next_action": "need_refine",
44
        "tool_call_count": state.get("tool_call_count", 0) + 1,
45
    }
46

47

48
async def refine_node(state: AgentState) -> dict:
49
    """优化回答节点：基于工具结果再次生成回复"""
50
    print("[优化节点] 基于工具结果优化回复...")
51
    response = await llm.ainvoke(state["messages"])
52
    return {"messages": [response], "next_action": "done"}
53

54

55
# 条件边函数：根据 State 返回下一个节点名称
56
# 返回值必须是 add_conditional_edges 中 path_map 的 key
57
def route_after_analyze(state: AgentState) -> Literal["direct_answer", "tool", "refine"]:
58
    """分析节点之后的路由"""
59
    action = state["next_action"]
60
    if action == "need_tool":
61
        return "tool"
62
    elif action == "need_refine":
63
        return "refine"
64
    else:
65
        return "direct_answer"
66

67

68
def route_after_tool(state: AgentState):
69
    """工具节点之后的路由：调用次数过多则强制结束，防止死循环"""
70
    if state.get("tool_call_count", 0) >= 3:
71
        return END
72
    return "refine"
73

74

75
async def main():
76
    builder = StateGraph(AgentState)
77
    builder.add_node("analyze", analyze_node)
78
    builder.add_node("direct_answer", direct_answer_node)
79
    builder.add_node("tool", tool_node)
80
    builder.add_node("refine", refine_node)
81
    builder.set_entry_point("analyze")
82

83
    # add_conditional_edges(起始节点, 路由函数, 路由映射)
84
    # 路由函数返回字符串，路由映射把字符串映射到实际节点名
85
    builder.add_conditional_edges(
86
        "analyze",
87
        route_after_analyze,
88
        {
89
            "direct_answer": "direct_answer",
90
            "tool": "tool",
91
            "refine": "refine",
92
        },
93
    )
94

95
    # 工具节点同样用条件边：超过 3 次强制结束，否则进入 refine
96
    builder.add_conditional_edges(
97
        "tool",
98
        route_after_tool,
99
        {"refine": "refine", END: END},
100
    )
101

102
    builder.add_edge("direct_answer", END)
103
    builder.add_edge("refine", END)
104

105
    graph = builder.compile()
106

107
    print("Graph 结构：")
108
    print(graph.get_graph().draw_mermaid())
109

110
    # 不同输入走不同路径
111
    test_cases = [
112
        "你好，介绍一下 Python",   # direct_answer 路径
113
        "武汉今天天气怎么样",       # tool → refine 路径
114
    ]
115

116
    for question in test_cases:
117
        print(f"\n{'=' * 50}")
118
        print(f"问题：{question}")
119
        result = await graph.ainvoke({
120
            "messages": [HumanMessage(content=question)],
121
            "next_action": "",
122
            "tool_call_count": 0,
123
        })
124
        last_msg = result["messages"][-1]
125
        print(f"最终回复：{last_msg.content[:100]}")
126

127

128
asyncio.run(main())

1
uv run python scripts/ch10/test_conditional_graph.py

1
import asyncio
2
from typing import TypedDict, Annotated
3
from langgraph.graph import StateGraph, END
4
import operator
5

6

7
class DemoState(TypedDict):
8
    # 追加模式：每次返回的值追加到列表
9
    log: Annotated[list[str], operator.add]
10
    # 覆盖模式：每次返回的值覆盖旧值
11
    counter: int
12
    last_node: str
13

14

15
def node_a(state: DemoState) -> dict:
16
    print(f"Node A 执行，当前 counter={state['counter']}")
17
    return {
18
        "log": ["Node A 执行了"],                       # 追加
19
        "counter": state["counter"] + 1,                # 覆盖
20
        "last_node": "A",
21
    }
22

23

24
def node_b(state: DemoState) -> dict:
25
    print(f"Node B 执行，当前 counter={state['counter']}")
26
    return {
27
        "log": ["Node B 执行了"],                       # 追加
28
        "counter": state["counter"] + 10,              # 覆盖
29
        "last_node": "B",
30
    }
31

32

33
async def main():
34
    builder = StateGraph(DemoState)
35
    builder.add_node("a", node_a)
36
    builder.add_node("b", node_b)
37
    builder.set_entry_point("a")
38
    builder.add_edge("a", "b")
39
    builder.add_edge("b", END)
40
    graph = builder.compile()
41

42
    result = await graph.ainvoke({
43
        "log": ["初始日志"],
44
        "counter": 0,
45
        "last_node": "",
46
    })
47

48
    print("\n最终结果：")
49
    print(f"log（追加，保留全部历史）：{result['log']}")
50
    print(f"counter（覆盖，只留最新值）：{result['counter']}")
51
    print(f"last_node（覆盖）：{result['last_node']}")
52

53

54
asyncio.run(main())

1
uv run python scripts/ch10/test_state_update.py

1
from typing import TypedDict, Annotated
2
import operator
3

4
from langchain_core.messages import BaseMessage, HumanMessage
5
from langgraph.graph import StateGraph, END
6

7
from app.ai.llm import llm
8

9

10
class ChatState(TypedDict):
11
    # 对话消息：追加模式，保留完整历史
12
    messages: Annotated[list[BaseMessage], operator.add]
13

14

15
async def chat_node(state: ChatState) -> dict:
16
    """对话节点：调用 LLM 生成回复"""
17
    response = await llm.ainvoke(state["messages"])
18
    return {"messages": [response]}
19

20

21
def build_chat_graph():
22
    builder = StateGraph(ChatState)
23
    builder.add_node("chat", chat_node)
24
    builder.set_entry_point("chat")
25
    builder.add_edge("chat", END)
26
    # 编译一次即可，后续全局复用
27
    return builder.compile()
28

29

30
chat_graph = build_chat_graph()
31

32

33
async def chat_once(question: str) -> str:
34
    """对外暴露的简单调用接口"""
35
    result = await chat_graph.ainvoke({
36
        "messages": [HumanMessage(content=question)]
37
    })
38
    return result["messages"][-1].content

1
import asyncio
2
from app.ai.graphs.chat_graph import chat_once
3

4

5
async def main():
6
    answer = await chat_once("用一句话解释什么是状态图")
7
    print(answer)
8

9

10
asyncio.run(main())