什么是LangGraph

如果说LangChain是一条笔直的高速公路，从入口到出口，中途存在加油站（Tools），不存在其他路线，那么LangGraph就是一条存在多条路线的高速公路，支持不同的路线，不同的选择，但支持最终同样的终点。

LangGraph与LangChain最大的不同就是，LangGraph支持多种条件选择，支持自我回环检测修复机制，以ReAct模式来运行输出。

业务场景

• AI 编辑器：与用户多轮对话实现AI Coder，例如：Trae、Cursor、Kiro等等
• 智能电商客服：通过外接RAG服务+多轮对话检测，结合用户行为意图提供不同操作
• 报告总结助手：多 Agent 协作+自我回环检测机制生成长文本报告，例如：法律、合同、解决方案等
• HR 招聘入职助手：招聘筛选 → 面试安排 → 背调 → Offer 生成 → 入职引导 → 试用期跟踪 一整套链路
• xxx等等

以上场景，如果单纯依靠LangChain完成，难度很大，但是使用LangGraph这种天然的工作流AI 框架将会变得非常容易。

但是!!! LangGraph 也并不是做这种工作流框架的唯一选择。据我了解，大部分公司或大公司都在选择自研类似 LangGraph 框架，所以学习 LangGraph 不是学习框架本身，而是学习这种设计思想

核心准备

在阅读本篇文章前，希望你已经有基础的LangChain技能知识储备，以及准备好了你的大模型apiKey和baseUrl等信息。
你可以考虑选择百炼、讯飞、火山、google、亚马逊、ollama等等平台来获取。

基础组件

LangGraph总体就只有四个大的组件：

• state：整个LangGraph的内存数据共享容器，这是一个实体类，实体类中存放我们的所有信息，这个信息能被不同的 Agent 和 node 获取和增删改
• node：工作流的节点，每一个节点都代表了一种处理数据的规则，例如：报告生成节点，数据处理节点，爬虫节点等等
• edge：边，通过边来连接节点，使得不同的节点能够相互流转，数据通过上方的 state 来共享获取
• graph：图，整个LangGraph的工作流，就是通过上方的 state+node+edge = graph

举个简单的例子：现在你需要开发一个个人小助手，提供两种对外交互的模式：

• 普通模式：通过与用户直接交流回复输出
• 技术专家模式：对外提供专业的技术分析，区别于普通输出的是，技术专家能够从不同的知识维度和法律维度提供帮助信息。

那么上述这个例子就能产生以下组件：

• 普通节点 1：主要作用是直接与大模型交互给用户回复输出
• 专家节点 2：外接 RAG，MCP ，爬虫等技能来获取更加专业的输出
• 用户行为分析节点 3：区分用户的查询意图是普通查询还是技术查询。
• 条件边：用来连接意图识别节点路由到普通节点还是专家节点上，不同的节点连接拥有不同的条件边

最终产生的工作流就如下：

快速入门

通过一个 demo 来了解如何做到快速入门案例

状态定义

LangGraph官网明确描述要求，state的定义必须符合 TypedDict 的要求，所以我们直接通过继承TypedDict的形式来定义状态：

from typing import TypedDict, Annotated, List
from langchain_core.messages import BaseMessage
from langgraph.graph import add_messages
class AgentState(TypedDict):
    messages: Annotated[List[BaseMessage], add_messages]
    intent: str  # 用户询问意图

messages：消息列表参数，类型为 BaseMessage，并且LangGraph提供add_messages方法自动合并返回的不同类型的消息
intent：用户意图字段，用来存放由用户行为意图节点分析得到的实际意图

节点定义

然后定义三个节点，分别是意图节点，普通节点和专家节点

# 节点 1 分析用户意图
def intent_classifier_node(state: AgentState):
    prompt = [
        SystemMessage(
            content="你是一个意图分类器。根据用户的消息，判断是以下哪种情况：\n1. 日常闲聊（回复：chat）\n2. 编程/技术问题（回复：technical）\n只回复英文单词，不要有其他内容。"),
        state["messages"][-1]
    ]
    response = llm.invoke(prompt)
    intent = response.content.strip().lower()
    print(f"--- 识别到的意图: {intent} ---")
    return {"intent": intent}


# ------------------------------------------
# 节点 2：闲聊模式
# ------------------------------------------
def chat_mode_node(state: AgentState):
    prompt = [
        SystemMessage(content="你是一个风趣幽默的陪聊机器人。请用轻松、搞笑的语气回复用户。"),
        *state["messages"]
    ]
    response = llm.invoke(prompt)
    return {"messages": [response]}


# ------------------------------------------
# 节点 3：技术专家模式
# ------------------------------------------
def technical_mode_node(state: AgentState):
    prompt = [
        SystemMessage(content="你是一个资深的全栈架构师。请用严谨、专业的语气回复用户，必要时给出代码示例。"),
        *state["messages"]
    ]
    response = llm.invoke(prompt)
    return {"messages": [response]}

路由函数

路由函数共享 state，所以能直接从 state 中拿到我们存放进去的意图

def route_after_intent(state: AgentState) -> Literal["chat_mode_node", "technical_mode_node"]:
    if state["intent"] == "technical":
        return "technical_mode_node"
    else:
        return "chat_mode_node"

组装 graph

紧接着，就能将上述中定义好的节点通过边来进行组装

# 定义图
workflow = StateGraph(AgentState)

# 添加节点 名称与函数名重名
workflow.add_node(intent_classifier_node)
workflow.add_node(chat_mode_node)
workflow.add_node(technical_mode_node)

# 添加边
workflow.add_edge(START, "intent_classifier_node")

# 添加条件边判断
workflow.add_conditional_edges(
    "intent_classifier_node",  # 起始节点
    route_after_intent,  # 路由判断
    {
        "chat_mode_node": "chat_mode_node",
        "technical_mode_node": "technical_mode_node"
    }  # 目标节点映射
)

# 两个节点都走向 end
workflow.add_edge("chat_mode_node", END)
workflow.add_edge("technical_mode_node", END)
# 编译图
app = workflow.compile()

这样我们就得到了刚才的工作流

测试输出

if __name__ == '__main__':
    graph = app.get_graph()
    # 打印 Mermaid 格式的文本 (可以复制到 mermaid.live 去看)
    print("\n" + "=" * 30)
    print("Graph Mermaid Code:")
    print("=" * 30)
    print(graph.draw_mermaid())

    print("--- 测试 1：说句闲话 ---")
    res1 = app.invoke({"messages": [HumanMessage(content="今天天气好无聊，讲个笑话吧")]})
    print("AI:", res1["messages"][-1].content)

    print("\n--- 测试 2：问个技术问题 ---")
    res2 = app.invoke({"messages": [HumanMessage(content="Python 里列表和元组有啥区别？")]})
    print("AI:", res2["messages"][-1].content)

所有代码如下：

from typing import TypedDict, Annotated, List, Literal

from langchain_core.messages import BaseMessage, HumanMessage, SystemMessage
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.constants import START, END
from langgraph.graph import add_messages, StateGraph


# 定义图的内存 add_messages 会自动添加各种角色类型的消息到内存中
class AgentState(TypedDict):
    messages: Annotated[List[BaseMessage], add_messages]
    intent: str  # 用户询问意图


from config.qwen_model import qwen_model

llm = qwen_model.get_llm_model()


# 节点 1 分析用户意图
def intent_classifier_node(state: AgentState):
    prompt = [
        SystemMessage(
            content="你是一个意图分类器。根据用户的消息，判断是以下哪种情况：\n1. 日常闲聊（回复：chat）\n2. 编程/技术问题（回复：technical）\n只回复英文单词，不要有其他内容。"),
        state["messages"][-1]
    ]
    response = llm.invoke(prompt)
    intent = response.content.strip().lower()
    print(f"--- 识别到的意图: {intent} ---")
    return {"intent": intent}


# ------------------------------------------
# 节点 2：闲聊模式
# ------------------------------------------
def chat_mode_node(state: AgentState):
    prompt = [
        SystemMessage(content="你是一个风趣幽默的陪聊机器人。请用轻松、搞笑的语气回复用户。"),
        *state["messages"]
    ]
    response = llm.invoke(prompt)
    return {"messages": [response]}


# ------------------------------------------
# 节点 3：技术专家模式
# ------------------------------------------
def technical_mode_node(state: AgentState):
    prompt = [
        SystemMessage(content="你是一个资深的全栈架构师。请用严谨、专业的语气回复用户，必要时给出代码示例。"),
        *state["messages"]
    ]
    response = llm.invoke(prompt)
    return {"messages": [response]}


# ------------------------------------------
# 【关键】路由函数：决定下一步去哪
# ------------------------------------------
def route_after_intent(state: AgentState) -> Literal["chat_mode_node", "technical_mode_node"]:
    if state["intent"] == "technical":
        return "technical_mode_node"
    else:
        return "chat_mode_node"


# 定义图
workflow = StateGraph(AgentState)

# 添加节点 名称与函数名重名
workflow.add_node(intent_classifier_node)
workflow.add_node(chat_mode_node)
workflow.add_node(technical_mode_node)

# 添加边
workflow.add_edge(START, "intent_classifier_node")

# 添加条件边判断
workflow.add_conditional_edges(
    "intent_classifier_node",  # 起始节点
    route_after_intent,  # 路由判断
    {
        "chat_mode_node": "chat_mode_node",
        "technical_mode_node": "technical_mode_node"
    }  # 目标节点映射
)

# 两个节点都走向 end
workflow.add_edge("chat_mode_node", END)
workflow.add_edge("technical_mode_node", END)

# 添加内存会话记忆
memory_saver = InMemorySaver()
# 编译图
app = workflow.compile(checkpointer=memory_saver)

if __name__ == '__main__':
    graph = app.get_graph()
    # 打印 Mermaid 格式的文本 (可以复制到 mermaid.live 去看)
    print("\n" + "=" * 30)
    print("Graph Mermaid Code:")
    print("=" * 30)
    print(graph.draw_mermaid())
    # 用户的 sessionId
    config = {"configurable": {"thread_id": "user_session_002"}}

    print("--- 测试 1：说句闲话 ---")
    res1 = app.invoke({"messages": [HumanMessage(content="今天天气好无聊，讲个笑话吧")]}, config=config)
    print("AI:", res1["messages"][-1].content)

    print("\n--- 测试 2：问个技术问题 ---")
    res2 = app.invoke({"messages": [HumanMessage(content="Python 里列表和元组有啥区别？")]}, config=config)
    print("AI:", res2["messages"][-1].content)