1. 概述#

本博客内容主要介绍如何将大模型集成到LangGraph框架中，构建一个有应用价值的问答流程。通过学习，我们可以：

了解如何在LangGraph中集成大语言模型
掌握构建问答流程的完整步骤
学习如何扩展图结构以实现更复杂的功能

2. 大模型接入方式#

2.1. 模型支持#

LangGraph对目前主流的在线或开源模型均支持接入，包括：

在线模型：OpenAI GPT系列、Anthropic Claude等
开源模型：GLM4、LLaMA3、Qwen等

2.2. 接入方式#

可以通过以下方式将大模型集成到LangGraph中：

使用LangChain框架：通过LangChain提供的接口（如ChatOpenAI）接入
原生API方式：直接使用大模型的原生API接口
推理加速库：利用Ollama、vLLM等推理加速库实现高效集成

参考文档：

LangChain Chat集成：https://python.langchain.com/docs/integrations/chat/

LangChain LLM集成：https://python.langchain.com/docs/integrations/llms/

3. 基础问答流程构建#

3.1. 定义图模式#

首先需要定义图的状态模式，包括输入状态、输出状态和整体状态：

1
from langgraph.graph import StateGraph
2
from typing_extensions import TypedDict
3
from langgraph.graph import START, END
4

5
# 定义输入的模式
6
class InputState(TypedDict):
7
    question: str
8

9
# 定义输出的模式
10
class OutputState(TypedDict):
11
    answer: str
12

13
# 将 InputState 和 OutputState 这两个 TypedDict 类型合并成一个更全面的字典类型
14
class OverallState(InputState, OutputState):
15
    pass

关键点说明：

InputState：定义用户输入的格式，包含question字段
OutputState：定义系统输出的格式，包含answer字段
OverallState：通过多重继承合并输入和输出状态，形成完整的状态定义

3.2. 创建LLM节点#

创建一个节点函数，用于接收用户问题并调用大模型生成回答：

1
from langchain_openai import ChatOpenAI
2
from langchain_core.prompts import ChatPromptTemplate
3
import getpass
4
import os
5

6
# 设置API密钥
7
if not os.environ.get("OPENAI_API_KEY"):
8
    os.environ["OPENAI_API_KEY"] = getpass.getpass("Enter your OpenAI API key: ")
9

10
def llm_node(state: InputState):
11
    messages = [
12
        ("system", "你是一位乐于助人的智能小助理"),
13
        ("human", state["question"])
14
    ]
15

16
    llm = ChatOpenAI(model="gpt-4o", temperature=0)
17
    response = llm.invoke(messages)
18

19
    return {"answer": response.content}

节点函数特点：

输入参数：接收state: InputState作为第一个位置参数
消息构建：使用系统提示词和用户问题构建消息列表
模型调用：使用ChatOpenAI调用GPT-4o模型
返回值：返回包含answer字段的字典，更新图状态

3.3. 构建图结构#

将节点添加到图中，并建立节点间的连接关系：

1
# 明确指定它的输入和输出数据的结构或模式
2
builder = StateGraph(OverallState, input=InputState, output=OutputState)
3

4
# 添加节点
5
builder.add_node("llm_node", llm_node)
6

7
# 添加边
8
builder.add_edge(START, "llm_node")
9
builder.add_edge("llm_node", END)
10

11
# 编译图
12
graph = builder.compile()

图结构说明：

StateGraph初始化时指定了状态模式、输入模式和输出模式
add_node方法添加节点，第一个参数是节点名称，第二个参数是节点函数
add_edge方法添加边，连接START → llm_node → END
compile方法编译图，验证图结构的合法性

3.4. 执行测试#

使用invoke方法调用图，传入用户问题：

1
# 基础测试
2
result = graph.invoke({"question": "你好，我用来测试"})
3
print(result["answer"])
4

5
# 详细问题测试
6
result = graph.invoke({"question": "你好，请你详细的介绍一下你自己"})
7
print(result["answer"])

执行流程：

用户输入通过START节点进入图
llm_node节点接收问题，调用大模型生成回答
结果通过END节点返回给用户

4. 扩展：多节点问答流程#

4.1. 扩展状态定义#

为了支持多节点间的数据传递，需要扩展状态定义：

1
from typing_extensions import TypedDict, Optional
2

3
# 定义输入的模式
4
class InputState(TypedDict):
5
    question: str
6
    llm_answer: Optional[str]  # 表示 llm_answer 可以是 str 类型，也可以是 None
7

8
# 定义输出的模式
9
class OutputState(TypedDict):
10
    answer: str
11

12
# 将 InputState 和 OutputState 这两个 TypedDict 类型合并成一个更全面的字典类型
13
class OverallState(InputState, OutputState):
14
    pass

关键变化：

添加了llm_answer字段，用于存储中间节点的输出结果
使用Optional[str]表示该字段可以为None，提供灵活性

4.2. 创建多个节点#

定义两个节点：一个用于生成回答，另一个用于翻译：

1
def llm_node(state: InputState):
2
    messages = [
3
        ("system", "你是一位乐于助人的智能小助理"),
4
        ("human", state["question"])
5
    ]
6

7
    llm = ChatOpenAI(model="gpt-4o", temperature=0)
8
    response = llm.invoke(messages)
9

10
    return {"llm_answer": response.content}
11

12
def action_node(state: InputState):
13
    messages = [
14
        ("system", "无论你接收到什么语言的文本，请翻译成法语"),
15
        ("human", state["llm_answer"])
16
    ]
17

18
    llm = ChatOpenAI(model="gpt-4o", temperature=0)
19
    response = llm.invoke(messages)
20

21
    return {"answer": response.content}

节点功能说明：

llm_node：接收用户问题，生成中文回答，并将结果存储在llm_answer字段
action_node：接收llm_answer，将其翻译成法语，并将结果存储在answer字段

4.3. 构建多节点图结构#

将两个节点添加到图中，并建立连接：

1
# 明确指定它的输入和输出数据的结构或模式
2
builder = StateGraph(OverallState, input=InputState, output=OutputState)
3

4
# 添加节点
5
builder.add_node("llm_node", llm_node)
6
builder.add_node("action_node", action_node)
7

8
# 添加边
9
builder.add_edge(START, "llm_node")
10
builder.add_edge("llm_node", "action_node")
11
builder.add_edge("action_node", END)
12

13
# 编译图
14
graph = builder.compile()

图结构流程：

1
START → llm_node → action_node → END

执行流程：

用户问题通过START进入llm_node
llm_node生成中文回答，更新llm_answer字段
action_node读取llm_answer，翻译成法语，更新answer字段
最终结果通过END返回

4.4. 测试多节点流程#

1
# 测试自我介绍翻译
2
result = graph.invoke({"question": "你好，请你详细的介绍一下你自己"})
3
print(result["answer"])
4

5
# 测试知识问答翻译
6
result = graph.invoke({"question": "请问什么是人工智能？"})
7
print(result["answer"])

5. 实测代码(基于Deepseek)#

5.1. 单Agent流程#

1
from langgraph.graph import StateGraph
2
from typing_extensions import TypedDict
3
from langgraph.graph import START, END
4

5
# 1. 创建图模式
6
# 为什么参数是TypedDict？
7
# TypedDict是Python中的一个类型，用于定义字典的类型。
8
# 它可以帮助我们定义字典的键和值的类型，从而避免在运行时出现类型错误。
9
class InputState(TypedDict):
10
  question: str
11

12
class OutputState(TypedDict):
13
  answer: str
14

15
# pass 和 ... 的区别
16
# pass 是一个占位符，表示什么也不做。
17
# ... 表示一个空字典。
18
class OverallState(InputState, OutputState):
19
  pass
20

21
# 2. 创建节点函数
22
# 获取api key
23
import os
24
from langchain_openai import ChatOpenAI
25
from langchain_core.prompts import ChatPromptTemplate
26
from langchain_core.messages import SystemMessage, HumanMessage
27

28
# 设置api key，这边直接配置了key，要科学保密可以使用dotenv来进行配置
29
os.environ["OPENAI_API_KEY"] = "sk-xxx" # 填自己的api-key
30

31

32
def llm_node(state: OverallState) -> OverallState:
33
  messages = [
34
    SystemMessage(content="你是一位乐于助人的智能小助理"),
35
    HumanMessage(content=state["question"])
36
  ]
37
  llm = ChatOpenAI(
38
    model="deepseek-chat",
39
    temperature=0,
40
    base_url="https://api.deepseek.com/v1"
41
  )
42
  response = llm.invoke(messages)
43
  return {"answer": response.content}
44

45
# 3. 创建图结构
46
builder = StateGraph(OverallState, input_schema=InputState, output_schema=OutputState)
47
builder.add_node("llm_node", llm_node)
48
builder.add_edge(START, "llm_node")
49
builder.add_edge("llm_node", END)
50

51
# 4. 编译图
52
graph = builder.compile()
53

54
# 5. 执行图
55
result = graph.invoke({"question": "你好，我用来测试"})
56
print(result["answer"])

运行截图：

5.2. 多Agent流程#

1
import os
2
from typing import Optional
3
from typing_extensions import TypedDict
4
from langchain_openai import ChatOpenAI
5
from langchain_core.messages import SystemMessage, HumanMessage
6
from langgraph.graph import StateGraph, START, END
7

8
class InputState(TypedDict):
9
  question: str
10
  llm_answer: Optional[str]
11

12
class OutputState(TypedDict):
13
  answer: str
14

15
class OverallState(InputState, OutputState):
16
  pass
17

18
os.environ["OPENAI_API_KEY"] = "sk-xxx" # 修改为自己的api-key
19

20
def llm_node(state: OverallState) -> OverallState:
21
  messages = [
22
    SystemMessage(content="你是一位乐于助人的智能小助理"),
23
    HumanMessage(content=state["question"])
24
  ]
25
  llm = ChatOpenAI(
26
    model="deepseek-chat",
27
    temperature=0,
28
    base_url="https://api.deepseek.com/v1"
29
  )
30
  response = llm.invoke(messages)
31
  return {"llm_answer": response.content}
32

33
def action_node(state: OverallState) -> OverallState:
34
  messages = [
35
    SystemMessage(content="请将你接收到的文本翻译为用英文"),
36
    HumanMessage(content=state["llm_answer"])
37
  ]
38
  llm = ChatOpenAI(
39
    model="deepseek-chat",
40
    temperature=0,
41
    base_url="https://api.deepseek.com/v1"
42
  )
43
  response = llm.invoke(messages)
44
  return {"answer": response.content}
45

46
builder = StateGraph(OverallState, input_schema=InputState, output_schema=OutputState)
47
builder.add_node("llm_node", llm_node)
48
builder.add_node("action_node", action_node)
49
builder.add_edge(START, "llm_node")
50
builder.add_edge("llm_node", "action_node")
51
builder.add_edge("action_node", END)
52
graph = builder.compile()
53
result = graph.invoke({"question": "你好，请你详细的介绍一下你自己"})
54
print(result["answer"])

运行截图：多Agent

6. 核心要点总结#

6.1. LangGraph的灵活性#

节点函数自主性：
- 可以在节点函数中定义任意逻辑
- 可以自主构建中间状态信息
- 不强制依赖LangChain框架
状态管理：
- 节点可以读取和写入图状态中的任何字段
- 状态在节点间自动传递和更新
- 通过TypedDict确保类型安全
图结构扩展性：
- 可以轻松添加新节点
- 通过边灵活控制执行流程
- 支持复杂的多节点工作流

6.2. 与LangChain的关系#

基于LangChain：LangGraph基于LangChain的表达式语言构建
可独立运行：虽然基于LangChain，但完全可以脱离LangChain独立运行
模型接入灵活：可以使用LangChain接口，也可以使用原生API

6.3. 实践建议#

理解状态模式：深入理解TypedDict的作用和状态传递机制
节点函数设计：合理设计节点函数，确保输入输出格式正确
图结构规划：在构建图之前，先规划好节点和边的连接关系
逐步扩展：从简单流程开始，逐步添加复杂功能

7. 学习要点#

通过本节的学习，需要掌握：

✅ 如何在LangGraph中定义图的状态模式
✅ 如何创建节点函数并集成大语言模型
✅ 如何构建图结构并建立节点间的连接
✅ 如何执行图并获取结果
✅ 如何扩展图结构以实现多节点工作流
✅ 理解节点间状态传递的机制

8. 下一步学习方向#

掌握了基础的问答流程构建后，可以进一步学习：

条件边（Conditional Edges）：根据条件动态选择执行路径
循环图（Cyclic Graphs）：实现循环和迭代逻辑
状态持久化：保存和恢复图执行状态
人机交互：在图中插入人工审核节点
流式输出：实现实时流式响应

注意：本节示例虽然不复杂，但涉及的知识点和细节较多。建议亲自尝试和体验，打好扎实的基础，才能更好地开展后续复杂循环图的学习。

Hygen's Blog