02_第一个LangGraph应用

发表于 更新于 分类于 ， Changyan： 本文字数： 3k 阅读时长 ≈ 11 分钟

动手实践！本节课将带你配置开发环境，使用通义千问API，编写第一个LangGraph程序。通过一个简单的Hello World示例，理解LangGraph代码的基本结构和运行流程。本课结束后，你将能够独立运行和调试LangGraph应用。

🎯 学习目标

学完本节课，你将能够：

  1. 配置LangGraph开发环境
  2. 获取和配置通义千问API密钥
  3. 编写和运行第一个LangGraph程序
  4. 理解LangGraph代码的基本结构
  5. 掌握基本的调试技巧

1. 环境准备

1.1 检查Python版本

LangGraph需要Python 3.10或更高版本：

1
2
python --version
# 应该显示：Python 3.10.x 或更高

1.2 安装依赖包

在项目根目录创建 .venv 虚拟环境（如果还没有）：

1
2
3
4
5
6
7
cd /Users/t-yuhaiyang/PycharmProjects/demo

# 如果还没有虚拟环境，创建一个
python -m venv .venv

# 激活虚拟环境
source .venv/bin/activate

使用uv安装依赖（推荐，更快）：

1
2
3
4
5
6
7
8
# 安装核心依赖
uv pip install langgraph langchain langchain-community

# 安装通义千问支持
uv pip install dashscope

# 安装辅助工具
uv pip install python-dotenv pydantic

或者使用普通pip：

1
pip install langgraph langchain langchain-community dashscope python-dotenv pydantic

1.3 获取通义千问API密钥

步骤1：注册阿里云账号

访问：https://dashscope.aliyun.com/

步骤2：开通DashScope服务

  • 登录后，点击”开通DashScope”
  • 选择”按量付费”（有免费额度）

步骤3：创建API Key

  • 进入控制台
  • 点击”API Key管理”
  • 创建新的API Key
  • 复制并保存好这个Key！

步骤4：配置环境变量

在 demo_16 目录下创建 .env 文件：

1
2
3
4
5
6
7
# 通义千问API密钥
DASHSCOPE_API_KEY=sk-xxxxxxxxxxxxxxxxxxxxxxxx

# 可选：如果要用其他模型
# DEEPSEEK_API_KEY=your_deepseek_key
# OPENAI_API_KEY=your_openai_key
# OPENAI_BASE_URL=https://api.your-proxy.com/v1

安全提示：

  • ⚠️ 不要把API Key提交到git！
  • ⚠️ 确保 .env 已添加到 .gitignore

2. Hello World示例

2.1 最简单的例子

创建 01_hello_world.py：

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
"""
最简单的LangGraph示例：单节点问候程序
学习目标：理解图、节点、状态的基本概念
"""
from typing import TypedDict
from langgraph.graph import StateGraph, END


# 步骤1：定义状态
class State(TypedDict):
    """状态定义：在节点间传递的数据"""
    name: str      # 输入：用户名字
    message: str   # 输出：问候消息


# 步骤2：定义节点
def greet_node(state: State) -> State:
    """
    问候节点：生成问候消息
    
    参数:
        state: 当前状态，包含name字段
    
    返回:
        更新后的状态，包含message字段
    """
    name = state["name"]
    state["message"] = f"你好，{name}！欢迎学习LangGraph！"
    return state


# 步骤3：构建图
def create_graph() -> StateGraph:
    """创建并配置图"""
    # 创建图实例
    graph = StateGraph(State)
    
    # 添加节点
    graph.add_node("greet", greet_node)
    
    # 设置入口点
    graph.set_entry_point("greet")
    
    # 设置出口
    graph.add_edge("greet", END)
    
    return graph


# 步骤4：运行
def main() -> None:
    """主函数"""
    # 编译图
    app = create_graph().compile()
    
    # 调用图，传入初始状态
    result = app.invoke({"name": "小明"})
    
    # 打印结果
    print(result["message"])


if __name__ == "__main__":
    main()

2.2 运行程序

1
2
cd demo_16/lesson_02
python 01_hello_world.py

输出：

1
你好，小明！欢迎学习LangGraph！

2.3 代码解析

1️⃣ 定义状态（State）

1
2
3
class State(TypedDict):
    name: str      # 输入数据
    message: str   # 输出数据

理解：

  • TypedDict 提供类型提示，让代码更安全
  • 定义了在节点间传递的数据结构
  • 就像定义了一个快递包裹的格式

2️⃣ 定义节点（Node）

1
2
3
4
def greet_node(state: State) -> State:
    name = state["name"]           # 读取输入
    state["message"] = "你好..."   # 写入输出
    return state                    # 返回状态

理解：

  • 节点就是一个函数
  • 输入：当前状态
  • 输出：更新后的状态
  • 节点的职责：完成一件具体的事

3️⃣ 构建图（Graph）

1
2
3
4
graph = StateGraph(State)          # 创建图
graph.add_node("greet", greet_node)  # 添加节点
graph.set_entry_point("greet")     # 设置入口
graph.add_edge("greet", END)       # 设置出口

理解：

  • StateGraph(State) - 创建图，指定状态类型
  • add_node("名称", 函数) - 给节点起名字
  • set_entry_point - 从哪个节点开始
  • add_edge("起点", "终点") - 连接节点
  • END - 特殊标记，表示结束

4️⃣ 运行图

1
2
3
app = create_graph().compile()     # 编译图
result = app.invoke({"name": "小明"})  # 运行
print(result["message"])           # 获取结果

理解：

  • compile() - 把图编译成可执行的应用
  • invoke(初始状态) - 执行图，传入初始数据
  • 返回最终状态

3. 进阶示例：多节点链式调用

3.1 示例：问候+询问

创建 02_chain_nodes.py：

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
"""
多节点示例：链式调用
学习目标：理解节点间的数据传递
"""
from typing import TypedDict
from langgraph.graph import StateGraph, END


# 定义状态
class State(TypedDict):
    name: str
    greeting: str
    question: str


# 节点1：问候
def greet_node(state: State) -> State:
    """生成问候语"""
    state["greeting"] = f"你好，{state['name']}！"
    print(f"[节点1] 执行问候: {state['greeting']}")
    return state


# 节点2：询问
def ask_node(state: State) -> State:
    """生成询问"""
    state["question"] = "今天过得怎么样？"
    print(f"[节点2] 执行询问: {state['question']}")
    return state


# 构建图
def create_graph() -> StateGraph:
    graph = StateGraph(State)
    
    # 添加两个节点
    graph.add_node("greet", greet_node)
    graph.add_node("ask", ask_node)
    
    # 设置流程：greet → ask → END
    graph.set_entry_point("greet")
    graph.add_edge("greet", "ask")
    graph.add_edge("ask", END)
    
    return graph


def main() -> None:
    app = create_graph().compile()
    result = app.invoke({"name": "小红"})
    
    print("\n[最终结果]")
    print(f"问候: {result['greeting']}")
    print(f"询问: {result['question']}")


if __name__ == "__main__":
    main()

输出：

1
2
3
4
5
6
[节点1] 执行问候: 你好，小红！
[节点2] 执行询问: 今天过得怎么样？

[最终结果]
问候: 你好，小红！
询问: 今天过得怎么样？

理解：

  • 节点按顺序执行：greet → ask
  • state 在节点间自动传递
  • 每个节点都能访问和修改 state

4. 集成AI：使用通义千问

4.1 示例：AI问候助手

创建 03_with_llm.py：

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
"""
集成通义千问：AI问候助手
学习目标：理解如何在节点中使用LLM
"""
import os
from typing import TypedDict
from langgraph.graph import StateGraph, END
from langchain_community.llms import Tongyi
from dotenv import load_dotenv

# 加载环境变量
load_dotenv()


# 定义状态
class State(TypedDict):
    user_name: str
    user_mood: str
    ai_response: str


# 初始化通义千问
def get_llm() -> Tongyi:
    """
    创建通义千问LLM实例
    
    注意：需要在.env中配置DASHSCOPE_API_KEY
    """
    api_key = os.getenv("DASHSCOPE_API_KEY")
    if not api_key:
        raise ValueError("请在.env文件中设置DASHSCOPE_API_KEY")
    
    return Tongyi(
        model="qwen-turbo",  # 使用qwen-turbo模型（便宜快速）
        api_key=api_key,
        temperature=0.7,     # 创意度：0-1，越高越有创意
    )


# 节点：AI生成个性化问候
def ai_greet_node(state: State) -> State:
    """
    使用AI生成个性化问候
    
    根据用户名字和心情，生成贴心的问候语
    """
    llm = get_llm()
    
    # 构造提示词
    prompt = f"""
    你是一个热情友好的助手。
    
    用户信息：
    - 名字：{state['user_name']}
    - 心情：{state['user_mood']}
    
    请生成一句贴心的问候语，要体现出对用户心情的关注。
    直接输出问候语，不要有其他多余内容。
    """
    
    # 调用LLM
    print("[AI思考中...]")
    response = llm.invoke(prompt)
    
    state["ai_response"] = response
    return state


# 构建图
def create_graph() -> StateGraph:
    graph = StateGraph(State)
    graph.add_node("ai_greet", ai_greet_node)
    graph.set_entry_point("ai_greet")
    graph.add_edge("ai_greet", END)
    return graph


def main() -> None:
    """运行示例"""
    app = create_graph().compile()
    
    # 测试不同场景
    test_cases = [
        {"user_name": "小明", "user_mood": "开心"},
        {"user_name": "小红", "user_mood": "有点累"},
        {"user_name": "小刚", "user_mood": "兴奋"},
    ]
    
    for case in test_cases:
        print(f"\n{'='*50}")
        print(f"输入: {case}")
        print(f"{'='*50}")
        
        result = app.invoke(case)
        print(f"\nAI回复: {result['ai_response']}")


if __name__ == "__main__":
    main()

4.2 运行AI示例

1
2
# 确保.env文件中有DASHSCOPE_API_KEY
python 03_with_llm.py

可能的输出：

1
2
3
4
5
6
7
8
9
10
11
12
13
==================================================
输入: {'user_name': '小明', 'user_mood': '开心'}
==================================================
[AI思考中...]

AI回复: 小明，你好呀！看到你这么开心，我也跟着高兴起来了！今天有什么开心的事要分享吗？

==================================================
输入: {'user_name': '小红', 'user_mood': '有点累'}
==================================================
[AI思考中...]

AI回复: 小红，辛苦了！感觉有点累的话，要记得好好休息哦。要不要喝杯热茶，放松一下？

4.3 代码要点

配置LLM

1
2
3
4
5
6
7
from langchain_community.llms import Tongyi

llm = Tongyi(
    model="qwen-turbo",      # 模型选择
    api_key=api_key,         # API密钥
    temperature=0.7,         # 创意度
)

模型选择：

  • qwen-turbo - 便宜快速，适合测试
  • qwen-plus - 更强能力，适合生产
  • qwen-max - 最强能力，适合复杂任务

temperature参数：

  • 0.0-0.3 - 稳定、一致（适合事实性任务）
  • 0.4-0.7 - 平衡（适合对话）
  • 0.8-1.0 - 创意、多样（适合创作）

调用LLM

1
2
3
response = llm.invoke(prompt)  # 同步调用
# 或
response = await llm.ainvoke(prompt)  # 异步调用（推荐）

5. 调试技巧

5.1 打印状态变化

在每个节点中打印状态：

1
2
3
4
5
6
7
8
def my_node(state: State) -> State:
    print(f"[进入节点] 当前状态: {state}")
    
    # 处理逻辑
    state["result"] = "处理完成"
    
    print(f"[离开节点] 更新后状态: {state}")
    return state

5.2 使用stream查看执行过程

1
2
3
4
5
app = create_graph().compile()

# 使用stream逐步查看执行
for step in app.stream({"name": "小明"}):
    print(f"当前步骤: {step}")

5.3 可视化图结构

1
2
3
4
5
from IPython.display import Image, display

# 在Jupyter中显示图
app = create_graph().compile()
display(Image(app.get_graph().draw_mermaid_png()))

5.4 异常处理

1
2
3
4
5
6
7
8
9
10
11
def safe_node(state: State) -> State:
    """带异常处理的节点"""
    try:
        # 可能出错的操作
        result = risky_operation(state)
        state["result"] = result
    except Exception as e:
        print(f"[错误] {e}")
        state["error"] = str(e)
    
    return state

6. 常见问题

Q1: 为什么要用TypedDict？

答： 提供类型提示，让IDE能够自动补全，减少错误：

1
2
3
4
5
6
7
8
# 用TypedDict
class State(TypedDict):
    name: str

def my_node(state: State) -> State:
    # IDE知道state有name字段，可以自动补全
    x = state["name"]  ✅
    y = state["age"]   ❌ IDE会提示错误

Q2: 节点一定要返回state吗？

答： 是的！节点必须返回state，否则数据无法传递到下一个节点：

1
2
3
4
5
6
7
8
9
# ❌ 错误：没有返回state
def bad_node(state: State):
    state["result"] = "done"
    # 忘记return了！

# ✅ 正确
def good_node(state: State) -> State:
    state["result"] = "done"
    return state

Q3: 可以修改state中不存在的字段吗？

答： 可以！TypedDict只是类型提示，运行时可以添加新字段：

1
2
3
4
5
6
7
class State(TypedDict):
    name: str

def my_node(state: State) -> State:
    # 虽然State定义中没有age，但运行时可以添加
    state["age"] = 18  # 可以，但不推荐
    return state

建议： 提前在State中定义所有可能用到的字段。

Q4: 通义千问调用失败怎么办？

排查步骤：

  1. 检查API Key是否正确

    1
    echo $DASHSCOPE_API_KEY

  2. 检查网络连接

    1
    ping dashscope.aliyuncs.com

  3. 检查余额
    登录DashScope控制台查看账户余额

  4. 查看错误信息

    1
    2
    3
    4
    5
    try:
        response = llm.invoke(prompt)
    except Exception as e:
        print(f"错误类型: {type(e)}")
        print(f"错误信息: {e}")

Q5: 节点执行顺序是怎样的？

答： 按照边的定义执行：

1
2
3
4
5
# 执行顺序：A → B → C
graph.set_entry_point("A")
graph.add_edge("A", "B")
graph.add_edge("B", "C")
graph.add_edge("C", END)

7. 练习题

练习1：个性化问候（基础）

修改 01_hello_world.py，增加时间判断：

  • 如果是早上（6-12点），说”早上好”
  • 如果是下午（12-18点），说”下午好”
  • 如果是晚上（18-24点），说”晚上好”
  • 其他时间说”深夜好”

提示：使用 datetime.now().hour

练习2：多轮对话（进阶）

创建一个三节点的对话链：

  1. 节点1：问”你叫什么名字？”
  2. 节点2：问”你来自哪里？”
  3. 节点3：根据前面的信息生成总结

练习3：AI翻译助手（AI集成）

创建一个翻译助手：

  • 输入：中文句子
  • 节点1：翻译成英文
  • 节点2：翻译成日文
  • 输出：中英日三种语言

提示：使用通义千问，prompt是关键！

练习4：错误处理（健壮性）

修改 03_with_llm.py，增加错误处理：

  • 如果API调用失败，重试3次
  • 如果3次都失败，返回默认问候语
  • 记录错误信息到state

8. 小结

你学到了什么

✅ 如何配置LangGraph开发环境
✅ 如何获取和使用通义千问API
✅ LangGraph代码的基本结构：

  • 定义State（状态）
  • 定义Node（节点函数）
  • 构建Graph（添加节点和边）
  • 运行Application（invoke调用）

✅ 如何在节点中集成LLM
✅ 基本的调试技巧

核心代码模板

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
# 1. 定义状态
class State(TypedDict):
    input: str
    output: str

# 2. 定义节点
def my_node(state: State) -> State:
    # 处理逻辑
    return state

# 3. 构建图
graph = StateGraph(State)
graph.add_node("my_node", my_node)
graph.set_entry_point("my_node")
graph.add_edge("my_node", END)

# 4. 运行
app = graph.compile()
result = app.invoke({"input": "test"})

📚 下一步

现在你已经会写基本的LangGraph程序了！但目前的例子都是简单的线性流程，实际应用中往往需要更复杂的状态管理。

下一课： Lesson 03: 状态管理和节点

我们将深入学习：

  • State的高级用法（列表、嵌套字典）
  • 节点间如何安全地共享数据
  • 状态更新的最佳实践
  • 构建一个多轮对话机器人

📖 参考资料

  • LangGraph快速开始
  • 通义千问API文档
  • Python TypedDict文档