langgraph-概念

一、定义

从 LangGraph 官方的定义看，该框架是一个用于使用大模型构建有状态、多参与者应用程序的库，可以创建代理和多代理工作流程。而其官方自己总结的 LangGraph 的优势则是：

• 循环和分支：在应用程序中实现循环和条件。
• 持久性：在图中的每个步骤之后自动保存状态。随时暂停和恢复图形执行，以支持错误恢复、人机交互工作流程等。
• 人机交互：中断图形执行以批准或编辑代理计划的下一个操作。
• 流支持：流输出由每个节点生成（包括令牌流）。
• 与 LangChain 集成：LangGraph 与 LangChain 和 LangSmith 无缝集成（但不需要它们）。

LangGraph Github：https://github.com/langchain-ai/langgraph
LangGraph Docs：https://langchain-ai.github.io/langgraph/
至此，当我们了解了上述的原理后，再来看 LangGraph 官方的介绍，就能够比较清楚的理解其独特优势究竟体现在何处。

二、LangGraph 底层源码解析

Graph 基类
对于任意一个简单或者复杂的图来说，都是基于 Graph 类来构建和管理图结构的。在 Graph 类中允许添加节点、边，并定义节点间的动态流转逻辑。如下是 Graph 类的主要组成部分和功能：
Class Graph : https://langchain-ai.github.io/langgraph/reference/graphs/#langgraph.graph.graph.Graph

一般构建图的时候，都用Graph的子类，GraphState，因为都带着状态。

Langgrph的State和CheckPoint类的区别：

State 是工作流运行时的“内存”，决定了数据如何在节点间传递和演变。
Checkpoint 则是“硬盘”，它把内存中的 State 以 thread_id 为维度持久化下来。
只用 LangState 确实能跑通单次工作流，但只有加上 Checkpoint，LangGraph 才能真正成为一个支持多轮对话、可中断、可恢复、可审计的生产级 AI Agent 框架。

from collections import defaultdict
from typing import Any, Callable, Dict, Optional, Set, Tuple, Union, Awaitable, Hashable

class Graph:
    def __init__(self) -> None:
        self.nodes: Dict[str, Any] = {}  # 一个字典，用于存储图中的所有节点。每个节点可以是一个字符串标识或者是一个可调用对象
        self.edges: Set[Tuple[str, str]] = set()  # 一个集合，用来存储图中所有的边，边由一对节点名称组成，表示从一个节点到另一个节点的直接连接。
        self.branches: defaultdict = defaultdict(dict)  # 一个默认字典，用于存储条件分支，允许从一个节点根据特定条件转移到多个不同的节点。
        self.support_multiple_edges = False  # 一个布尔值，指示图是否支持同一对节点间的多条边。
        self.compiled = False  # 一个布尔值，表示图是否已经被编译。编译是指图的结构已经设置完毕，准备进行执行。

    @property
    def _all_edges(self) -> Set[Tuple[str, str]]:
        """
        获取所有的边的信息。
        """
        return self.edges

    def add_node(self, node: Union[str, Callable], action: Optional[Callable] = None, *, metadata: Optional[Dict[str, Any]] = None) -> 'Graph':
        """
        添加一个新节点到图中。节点可以有附加的元数据，这些元数据存储在节点的字典中。
        """
        pass

    def add_edge(self, start_key: str, end_key: str) -> 'Graph':
        """
        在图中添加一条边，连接两个指定的节点。
        """
        pass

    def add_conditional_edges(self, source: str, path: Callable, path_map: Optional[Dict[Hashable, str]] = None, then: Optional[str] = None) -> 'Graph':
        """
        添加一个条件边，允许在执行时根据某个条件从一个节点动态地转移到一个或多个节点。
        """
        pass

    def set_entry_point(self, key: str) -> 'Graph':
        """
        设置图的入口点，即定义图执行的起始节点。
        """
        pass

    def set_conditional_entry_point(self, path: Callable, path_map: Optional[Dict[Hashable, str]] = None, then: Optional[str] = None) -> 'Graph':
        """
        设置一个条件入口点，允许根据条件动态决定图的起始执行点。
        """
        pass

    def set_finish_point(self, key: str) -> 'Graph':
        """
        设置结束点，定义图执行到此节点时将停止。
        """
        pass

    def validate(self, interrupt: Optional[Set[str]] = None) -> 'Graph':
        """
        验证图的结构是否正确，确保所有节点和边的定义都符合逻辑和图的规则。
        """
        pass

    def compile(self, checkpointer=None, interrupt_before: Optional[Set[str]] = None, interrupt_after: Optional[Set[str]] = None, debug: bool = False) -> 'Graph':
        """
        编译图，确认图的结构合法且可执行后，准备图以供执行。
        """
        pass

# 3.2 GraphState
# 定义图时要做的第一件事是定义图的 State。状态表示会随着图计算的进行而维护和更新的上下文或记忆。它用来确保图中的每个步骤都可以访问先前步骤的相关信息，从而可以根据整个过程中积累的数据进行动态决策。这个过程通过状态图 StateGraph 类实现，它继承自 Graph 类，这意味着 StateGraph 会使用或扩展基类的属性和方法。
# Class StateGraph: https://langchain-ai.github.io/langgraph/reference/graphs/#langgraph.graph.state.StateGraph

from collections import defaultdict
from typing import Any, Callable, Dict, Optional, Set, Tuple, Type, Union

class StateGraph(Graph):
    """StateGraph 是一个管理状态并通过定义的输入和输出架构支持状态转换的图。"""
    def __init__(self, state_schema: Optional[Type[Any]] = None, config_schema: Optional[Type[Any]] = None) -> None:
        super().__init__()
        self.state_schema = state_schema  # 一个可选的类型参数，定义图状态的结构。这是用于定义和验证图中节点处理的状态数据的模式。
        self.config_schema = config_schema  # 一个可选的类型参数，用于定义配置的结构。这可以用于定义和验证图的配置参数。
        input: Optional[Type[Any]] = None,  # 消息输入
        output: Optional[Type[Any]] = None, # 消息输出

    def add_node(self, node: Union[str, Callable], action: Optional[Callable] = None, *, metadata: Optional[Dict[str, Any]] = None) -> 'StateGraph':
        """向图中添加一个新节点。节点可以是一个具名字符串或一个可调用对象（如函数），如果node是字符串，则action应为与节点关联的可调用动作。"""
        pass

    def add_edge(self, start_key: str, end_key: str) -> 'StateGraph':
        """在图中添加一条边，连接两个节点。"""
        pass

    def compile(self) -> 'CompiledStateGraph':
        """编译图，将其转换成可运行的形式。包括验证图的完整性、预处理数据等。"""
        pass

"""
一、什么是图的模式
默认情况下，StateGraph 使用单模式运行，意味着图所有阶段读写同一个状态通道，全部节点共用该通道通信。
当存在以下需求时，需要自定义图模式：
内部节点需要传递输入 / 输出不需要的中间信息
图需要区分独立的输入、输出结构（例如输出仅返回单一结果字段）
LangGraph 底层通过 state_schema 灵活配置多种图模式；自定义复杂图模式涉及更多底层概念，将在课程后半段讲解。
单模式核心逻辑：所有模式均包含输入、输出两层校验
输入模式：校验外部传入数据是否匹配预设结构
输出模式：过滤内部状态冗余字段，仅对外返回指定数据
结构校验依赖 Python TypedDict 工具实现

"""

# TypedDict 介绍
# TypedDict 是 Python typing 模块提供的类型注解工具，可为字典的键强制约束字段类型，提升代码可读性与类型安全，适用于结构固定的字典场景。
# 在 LangGraph 中用于约束节点间流转的状态数据结构，避免字段缺失、类型错误导致流程异常。

from typing import TypedDict

class Contact(TypedDict):
    name: str
    email: str
    phone: str

def send_email(contact: Contact) -> None:
    print(f"Sending email to {contact['name']} at {contact['email']}")

# 实例化符合 Contact 结构的字典
contact_info: Contact = {
    'name': 'Lilei',
    'email': 'Lilei@qq.com',
    'phone': '15814023435'
}

send_email(contact_info)

# 示例说明：Contact 强制规定 name/email/phone 三个字符串必填字段，实例字典必须完整包含所有字段；函数接收时会做类型校验，适合 API 数据、模块间复杂状态传递场景。

# LangGraph 单输入输出模式实践代码
# ! pip install langgraph==0.2.60
from langgraph.graph import StateGraph
from typing_extensions import TypedDict

# 定义图外部输入结构
class InputState(TypedDict):
    question: str

# 定义图对外输出结构
class OutputState(TypedDict):
    answer: str

# 合并输入、输出字段，作为图完整内部状态
class OverallState(InputState, OutputState):
    pass

# 初始化 StateGraph

# 指定完整内部状态、独立输入输出模式
builder = StateGraph(OverallState, input=InputState, output=OutputState)

# OverallState：图运行全程存储的完整状态，包含输入、输出、内部中间字段
# input=InputState：约束外部调用图时仅需传入 question 字段
# output=OutputState：图执行结束后，仅对外返回 answer 字段，隐藏内部状态

# 3.3 Nodes（节点）
# 节点定义规则
# LangGraph 中节点是同步 / 异步 Python 函数，固定接收 state 作为第一个入参，读取当前状态执行业务逻辑，返回字典更新状态；无返回值代表不修改状态。

# Agent节点，无状态更新返回
def agent_node(state: InputState):
    print("我是一个AI Agent。")
    return

# 执行节点，返回字典更新state中的answer字段
def action_node(state: InputState):
    print("我现在是一个执行者。")
    return {"answer": "我现在执行成功了"}

# 将节点加入图
# 使用 add_node(节点名称, 函数) 注册节点，支持自定义节点标识；不指定名称时默认复用函数名。

builder.add_node("agent_node", agent_node)
builder.add_node("action_node", action_node)
# 当前图仅存在两个孤立节点，需要通过 Edges（边） 定义节点流转顺序。

"""3.4 Edges（边）
边的作用
定义节点间路由逻辑、图的起始与终止规则，是串联节点、实现流程调度的核心组件，分为 4 类：
普通边：固定单向流转，直接从 A 节点跳转至 B 节点
条件边：执行判断函数，动态决定下一跳节点（分支 / 循环）
入口点：图接收输入后首个执行节点
条件入口点：通过函数动态选择初始执行节点
特殊内置节点
START：图的输入起点，用户输入会传入该节点，用于指定流程首个节点
END：图终止终点，流转至此后流程结束，无后续操作"""

# 普通边完整串联示例
from langgraph.graph import START, END

# 从入口流转到agent节点
builder.add_edge(START, "agent_node")
# agent执行完流转到执行节点
builder.add_edge("agent_node", "action_node")
# 执行节点完成后结束流程
builder.add_edge("action_node", END)

# 编译图
# 调用 compile() 校验图结构（检测孤立节点、非法流转等问题），生成可执行图实例：
graph = builder.compile()
# 编译完成后，完整图结构构建完毕，可接收用户输入执行流程。

# Graph 的调用方法
# 核心调用方法：invoke()
# 编译完成的 graph 对象使用 invoke() 执行完整流程，入参必须严格匹配输入状态 InputState 的字典结构。

# 示例1：基础提问输入
graph.invoke({"question": "你好"})

# 示例2：业务场景提问输入
graph.invoke({"question": "今天的天气怎么样？"})

"""完整执行流程说明
外部传入 {"question": "xxx"}，匹配 InputState 输入规范；
状态数据沿边流转：START → agent_node → action_node → END；
节点间互通信息的底层原理：
所有节点共享全局 OverallState 状态容器；
任意节点均可读写状态内所有字段，实现跨节点数据传递；
InputState、OutputState 仅作为输入输出过滤器，对内完整状态无影响。
流程抵达 END 节点后自动终止，仅返回 OutputState 定义的输出字段（answer）。"""

三、使用 LangGraph 构建大模型的问答流程

在上面的示例中，我们通过使用打印函数来初步了解 LangGraph 构建图的基本方法和机制。接下来，我们将探索如何将大模型集成至 LangGraph 框架中，从而构建一个更具实际应用价值的用于问答流程的图模式。
首先，LangGraph 对目前主流的在线或者开源模型均支持接入，所以大家可以在该框架下非常便捷的应用到自己偏爱的大模型来进行问答流程的构建。这下面的示例中，我们选择比较方便且高效的 LangChain 框架，同时使用 OpenAI 的 GPT 模型来进行案例实现。而关于 LangChain 支持接入的模型列表及方式，大家可以在 LangChain Docs 中查阅：https://python.langchain.com/docs/integrations/chat/ 或者 https://python.langchain.com/docs/integrations/llms/。
这里仍然需要首先定义图模式，代码如下：

from langgraph.graph import StateGraph
from typing_extensions import TypedDict
from langgraph.graph import START, END

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

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

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

# 使用 OpenAI 的 GPT 模型需要使用到 ChatOpenAI 方法，我们需要将其定义到 Agent 节点中，用来接收用户输入的问题，调用 GPT 模型来根据用户的问题生成自然语言的回复响应。代码如下：
# 1、配置模型
import os
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
load_dotenv()
key = os.environ["OPENAI_API_KEY"]
base_url = os.environ["OPENAI_BASE_URL"]
model_name = os.environ["OPENAI_MODEL_NAME"]
llm = ChatOpenAI(model=model_name, api_key=key, base_url=base_url, temperature=0, request_timeout=10)

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate

import getpass
import os

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

    response = llm.invoke(messages)

    return {"answer": response.content}

# 构建图，添加节点和边，并进行图结构的编译。完整代码如下所示：

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

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

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

# 编译图
graph = builder.compile()

# 进行测试：

graph.invoke({"question":"你好，我用来测试"})

final_answer = graph.invoke({"question":"你好，我用来测试"})
print(final_answer["answer"])

final_answer = graph.invoke({"question":"你好，请你详细的介绍一下你自己"})
print(final_answer["answer"])

四、复杂流程实现

# 更进一步地，如果想在原有的图结构中构建更复杂的功能，则只需要新定义一个 Python 函数，并按照自己的预期流程用边来建立连接，如下代码所示：
from langgraph.graph import StateGraph
from typing_extensions import TypedDict, Optional
from langgraph.graph import START, END

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

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

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

# 我们定义了一个 action_node 节点，用来接收 llm_node 的输出，将其翻译成中文，如下代码所示：
# 1、配置模型
import os
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
load_dotenv()
key = os.environ["OPENAI_API_KEY"]
base_url = os.environ["OPENAI_BASE_URL"]
model_name = os.environ["OPENAI_MODEL_NAME"]
llm = ChatOpenAI(model=model_name, api_key=key, base_url=base_url, temperature=0, request_timeout=100)
def llm_node(state: InputState):
    messages = [
        ("system","你是一位乐于助人的智能小助理",),
        ("human", state["question"])
    ]
    response = llm.invoke(messages)

    return {"llm_answer": response.content}

def action_node(state: InputState):
    messages = [
        ("system","无论你接收到什么语言的文本，请翻译成英语",),
        ("human", state["llm_answer"])
    ]
    response = llm.invoke(messages)

    return {"answer": response.content}

# 构建图，添加节点和边，并进行图结构的编译。

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

# 添加节点
builder.add_node("llm_node", llm_node)
builder.add_node("action_node", action_node)

# 添加边
builder.add_edge(START, "llm_node")
builder.add_edge("llm_node", "action_node")
builder.add_edge("action_node", END)

# 编译图
graph = builder.compile()

final_answer = graph.invoke({"question":"你好，请你详细的介绍一下你自己"})
print(final_answer["answer"])