LangGraph完全指南

LangGraph是LangChain生态中用于构建复杂Agent工作流的框架，通过图结构和状态机实现可控的、可循环的AI应用。

什么是LangGraph

为什么需要LangGraph

传统的LangChain Chain是线性的、单向的：

python

1# LangChain Chain - 线性流程
2input -> chain1 -> chain2 -> chain3 -> output

但实际应用往往需要：

✅ 循环和条件分支
✅ 多Agent协作
✅ 复杂的决策流程
✅ 人工干预和审批
✅ 状态管理和持久化

LangGraph通过图结构解决这些问题：

python

1# LangGraph - 图状流程
2       ┌─────────┐
3       │  Input  │
4       └────┬────┘
5            │
6       ┌────▼────┐
7       │ Agent A │◄────┐
8       └────┬────┘     │
9            │          │
10        ┌───▼───┐      │
11        │ Judge │      │
12        └───┬───┘      │
13            │          │
14      ┌─────┴─────┐    │
15      │           │    │
16   ┌──▼──┐    ┌──▼──┐ │
17   │Done │    │Retry├─┘
18   └─────┘    └─────┘

核心概念

State (状态): 在图中流动的数据
Nodes (节点): 处理状态的函数
Edges (边): 连接节点的路径
Conditional Edges (条件边): 根据状态动态选择路径

快速开始

安装

bash

1pip install langgraph

Hello World

python

1from langgraph.graph import Graph
2from typing import TypedDict
3
4# 1. 定义状态
5class State(TypedDict):
6    message: str
7    count: int
8
9# 2. 定义节点函数
10def process_node(state: State) -> State:
11    state["count"] += 1
12    state["message"] = f"处理了 {state['count']} 次"
13    return state
14
15def check_node(state: State) -> State:
16    if state["count"] < 3:
17        state["message"] += " - 继续"
18    else:
19        state["message"] += " - 完成"
20    return state
21
22# 3. 构建图
23workflow = Graph()
24
25# 添加节点
26workflow.add_node("process", process_node)
27workflow.add_node("check", check_node)
28
29# 添加边
30workflow.add_edge("process", "check")
31
32# 添加条件边
33def should_continue(state: State) -> str:
34    if state["count"] < 3:
35        return "process"
36    return "end"
37
38workflow.add_conditional_edges(
39    "check",
40    should_continue,
41    {
42        "process": "process",
43        "end": END
44    }
45)
46
47# 设置入口
48workflow.set_entry_point("process")
49
50# 4. 编译并运行
51app = workflow.compile()
52
53result = app.invoke({"message": "", "count": 0})
54print(result)
55# {'message': '处理了 3 次 - 完成', 'count': 3}

核心组件详解

1. State (状态)

状态是图中流动的数据结构。

python

1from typing import TypedDict, Annotated
2from operator import add
3
4# 基础状态
5class BasicState(TypedDict):
6    input: str
7    output: str
8
9# 带类型注解的状态
10class AdvancedState(TypedDict):
11    # 普通字段（替换）
12    current_task: str
13    
14    # 可累加字段（追加）
15    messages: Annotated[list, add]
16    logs: Annotated[list, add]
17
18# 使用示例
19def node_a(state: AdvancedState) -> AdvancedState:
20    return {
21        "current_task": "任务A",
22        "messages": ["消息A"],  # 追加到现有列表
23        "logs": ["日志A"]
24    }
25
26def node_b(state: AdvancedState) -> AdvancedState:
27    return {
28        "current_task": "任务B",  # 替换
29        "messages": ["消息B"],  # 追加
30        "logs": ["日志B"]
31    }
32
33# 执行后: messages = ["消息A", "消息B"]

2. Nodes (节点)

节点是处理逻辑的载体。

python

1from langchain.chat_models import ChatOpenAI
2from langchain.schema import HumanMessage
3
4# 普通函数节点
5def simple_node(state: State) -> State:
6    state["value"] += 1
7    return state
8
9# LLM节点
10def llm_node(state: State) -> State:
11    llm = ChatOpenAI()
12    response = llm([HumanMessage(content=state["input"])])
13    state["output"] = response.content
14    return state
15
16# 异步节点
17async def async_node(state: State) -> State:
18    import asyncio
19    await asyncio.sleep(1)
20    state["processed"] = True
21    return state
22
23# 带错误处理的节点
24def safe_node(state: State) -> State:
25    try:
26        # 处理逻辑
27        result = risky_operation(state["data"])
28        state["result"] = result
29    except Exception as e:
30        state["error"] = str(e)
31    return state

3. Edges (边)

边定义节点间的连接。

python

1# 普通边 - 无条件连接
2workflow.add_edge("node_a", "node_b")
3
4# 条件边 - 根据状态选择路径
5def router(state: State) -> str:
6    if state["value"] > 10:
7        return "high_path"
8    else:
9        return "low_path"
10
11workflow.add_conditional_edges(
12    "decision_node",
13    router,
14    {
15        "high_path": "node_high",
16        "low_path": "node_low"
17    }
18)
19
20# 入口和出口
21workflow.set_entry_point("start_node")
22workflow.set_finish_point("end_node")

实战项目

项目1: 简单的对话Agent

python

1from langgraph.graph import Graph, END
2from langchain.chat_models import ChatOpenAI
3from typing import TypedDict, Annotated
4from operator import add
5
6class ConversationState(TypedDict):
7    messages: Annotated[list, add]
8    next_action: str
9
10# 对话节点
11def chat_node(state: ConversationState) -> ConversationState:
12    llm = ChatOpenAI(model="gpt-4")
13    
14    response = llm.invoke(state["messages"])
15    
16    return {
17        "messages": [response],
18        "next_action": "check"
19    }
20
21# 检查节点 - 判断是否继续
22def check_node(state: ConversationState) -> ConversationState:
23    last_message = state["messages"][-1].content.lower()
24    
25    if "再见" in last_message or "拜拜" in last_message:
26        state["next_action"] = "end"
27    else:
28        state["next_action"] = "continue"
29    
30    return state
31
32# 构建图
33workflow = Graph()
34
35workflow.add_node("chat", chat_node)
36workflow.add_node("check", check_node)
37
38workflow.add_edge("chat", "check")
39
40def router(state: ConversationState) -> str:
41    return state["next_action"]
42
43workflow.add_conditional_edges(
44    "check",
45    router,
46    {
47        "continue": "chat",
48        "end": END
49    }
50)
51
52workflow.set_entry_point("chat")
53
54app = workflow.compile()
55
56# 使用
57from langchain.schema import HumanMessage
58
59result = app.invoke({
60    "messages": [HumanMessage(content="你好")],
61    "next_action": ""
62})

项目2: 多Agent协作系统

python

1from langgraph.graph import Graph, END
2from langchain.chat_models import ChatOpenAI
3from typing import TypedDict, List
4
5class MultiAgentState(TypedDict):
6    task: str
7    research_result: str
8    code_result: str
9    review_result: str
10    final_output: str
11    current_agent: str
12
13# 研究Agent
14def research_agent(state: MultiAgentState) -> MultiAgentState:
15    llm = ChatOpenAI(model="gpt-4")
16    
17    prompt = f"""
18    作为研究专家，请分析以下任务：
19    {state['task']}
20    
21    提供技术方案和建议。
22    """
23    
24    response = llm.invoke([{"role": "user", "content": prompt}])
25    
26    return {
27        "research_result": response.content,
28        "current_agent": "coder"
29    }
30
31# 编码Agent
32def coder_agent(state: MultiAgentState) -> MultiAgentState:
33    llm = ChatOpenAI(model="gpt-4")
34    
35    prompt = f"""
36    基于以下研究结果编写代码：
37    {state['research_result']}
38    
39    任务：{state['task']}
40    """
41    
42    response = llm.invoke([{"role": "user", "content": prompt}])
43    
44    return {
45        "code_result": response.content,
46        "current_agent": "reviewer"
47    }
48
49# 审查Agent
50def reviewer_agent(state: MultiAgentState) -> MultiAgentState:
51    llm = ChatOpenAI(model="gpt-4")
52    
53    prompt = f"""
54    审查以下代码：
55    {state['code_result']}
56    
57    研究背景：{state['research_result']}
58    
59    如果代码有问题，返回"NEEDS_REVISION: 问题描述"
60    如果代码没问题，返回"APPROVED: 审查意见"
61    """
62    
63    response = llm.invoke([{"role": "user", "content": prompt}])
64    
65    return {
66        "review_result": response.content,
67        "current_agent": "router"
68    }
69
70# 构建图
71workflow = Graph()
72
73workflow.add_node("researcher", research_agent)
74workflow.add_node("coder", coder_agent)
75workflow.add_node("reviewer", reviewer_agent)
76
77workflow.add_edge("researcher", "coder")
78workflow.add_edge("coder", "reviewer")
79
80# 条件路由
81def review_router(state: MultiAgentState) -> str:
82    if "NEEDS_REVISION" in state["review_result"]:
83        return "coder"  # 返回重新编码
84    else:
85        return "end"  # 完成
86
87workflow.add_conditional_edges(
88    "reviewer",
89    review_router,
90    {
91        "coder": "coder",
92        "end": END
93    }
94)
95
96workflow.set_entry_point("researcher")
97
98app = workflow.compile()
99
100# 使用
101result = app.invoke({
102    "task": "实现一个快速排序算法",
103    "research_result": "",
104    "code_result": "",
105    "review_result": "",
106    "final_output": "",
107    "current_agent": ""
108})

项目3: RAG增强的问答系统

python

1from langgraph.graph import Graph, END
2from langchain.vectorstores import Chroma
3from langchain.embeddings import OpenAIEmbeddings
4from langchain.chat_models import ChatOpenAI
5from typing import TypedDict
6
7class RAGState(TypedDict):
8    question: str
9    documents: list
10    answer: str
11    needs_clarification: bool
12    clarification: str
13
14# 检索节点
15def retrieve_node(state: RAGState) -> RAGState:
16    vectorstore = Chroma(
17        persist_directory="./db",
18        embedding_function=OpenAIEmbeddings()
19    )
20    
21    docs = vectorstore.similarity_search(
22        state["question"],
23        k=5
24    )
25    
26    return {
27        "documents": [doc.page_content for doc in docs]
28    }
29
30# 判断节点 - 是否需要澄清
31def judge_node(state: RAGState) -> RAGState:
32    if not state["documents"]:
33        return {
34            "needs_clarification": True,
35            "clarification": "没有找到相关文档，请提供更多信息。"
36        }
37    
38    # 检查文档相关性
39    llm = ChatOpenAI()
40    check_prompt = f"""
41    问题：{state['question']}
42    文档：{state['documents'][0][:200]}...
43    
44    这些文档是否足够回答问题？回答YES或NO。
45    """
46    
47    response = llm.invoke([{"role": "user", "content": check_prompt}])
48    
49    if "NO" in response.content:
50        return {
51            "needs_clarification": True,
52            "clarification": "找到的信息不够完整，请补充更多细节。"
53        }
54    
55    return {"needs_clarification": False}
56
57# 生成节点
58def generate_node(state: RAGState) -> RAGState:
59    llm = ChatOpenAI(model="gpt-4")
60    
61    context = "\n\n".join(state["documents"])
62    
63    prompt = f"""
64    基于以下上下文回答问题：
65    
66    上下文：
67    {context}
68    
69    问题：{state['question']}
70    
71    请提供准确、详细的答案。
72    """
73    
74    response = llm.invoke([{"role": "user", "content": prompt}])
75    
76    return {"answer": response.content}
77
78# 构建图
79workflow = Graph()
80
81workflow.add_node("retrieve", retrieve_node)
82workflow.add_node("judge", judge_node)
83workflow.add_node("generate", generate_node)
84
85workflow.add_edge("retrieve", "judge")
86
87def judge_router(state: RAGState) -> str:
88    if state["needs_clarification"]:
89        return "end"
90    return "generate"
91
92workflow.add_conditional_edges(
93    "judge",
94    judge_router,
95    {
96        "generate": "generate",
97        "end": END
98    }
99)
100
101workflow.add_edge("generate", END)
102
103workflow.set_entry_point("retrieve")
104
105app = workflow.compile()

项目4: 人工审批流程

python

1from langgraph.graph import Graph, END
2from langgraph.checkpoint.sqlite import SqliteSaver
3from typing import TypedDict
4
5class ApprovalState(TypedDict):
6    request: str
7    analysis: str
8    risk_level: str
9    approved: bool
10    human_decision: str
11
12# 分析节点
13def analyze_node(state: ApprovalState) -> ApprovalState:
14    llm = ChatOpenAI(model="gpt-4")
15    
16    prompt = f"""
17    分析以下请求的风险等级（LOW/MEDIUM/HIGH）：
18    {state['request']}
19    
20    提供分析理由和风险评估。
21    """
22    
23    response = llm.invoke([{"role": "user", "content": prompt}])
24    analysis = response.content
25    
26    # 提取风险等级
27    if "HIGH" in analysis:
28        risk_level = "HIGH"
29    elif "MEDIUM" in analysis:
30        risk_level = "MEDIUM"
31    else:
32        risk_level = "LOW"
33    
34    return {
35        "analysis": analysis,
36        "risk_level": risk_level
37    }
38
39# 决策路由
40def decision_router(state: ApprovalState) -> str:
41    if state["risk_level"] == "LOW":
42        return "auto_approve"
43    else:
44        return "human_review"
45
46# 自动批准节点
47def auto_approve_node(state: ApprovalState) -> ApprovalState:
48    return {
49        "approved": True,
50        "human_decision": "自动批准（低风险）"
51    }
52
53# 人工审批节点 - 需要中断
54def human_review_node(state: ApprovalState) -> ApprovalState:
55    # 这里会中断，等待人工输入
56    return state
57
58# 构建图
59workflow = Graph()
60
61workflow.add_node("analyze", analyze_node)
62workflow.add_node("auto_approve", auto_approve_node)
63workflow.add_node("human_review", human_review_node)
64
65workflow.add_conditional_edges(
66    "analyze",
67    decision_router,
68    {
69        "auto_approve": "auto_approve",
70        "human_review": "human_review"
71    }
72)
73
74workflow.add_edge("auto_approve", END)
75workflow.add_edge("human_review", END)
76
77workflow.set_entry_point("analyze")
78
79# 使用checkpoint支持中断和恢复
80memory = SqliteSaver.from_conn_string(":memory:")
81app = workflow.compile(checkpointer=memory)
82
83# 运行
84config = {"configurable": {"thread_id": "1"}}
85
86# 第一次运行 - 可能会在human_review处中断
87result = app.invoke({
88    "request": "删除生产数据库",
89    "analysis": "",
90    "risk_level": "",
91    "approved": False,
92    "human_decision": ""
93}, config)
94
95# 人工决策后继续
96if result["risk_level"] in ["HIGH", "MEDIUM"]:
97    # 更新状态
98    app.update_state(
99        config,
100        {"approved": True, "human_decision": "人工批准"}
101    )
102    
103    # 继续执行
104    final_result = app.invoke(None, config)

高级特性

1. 持久化和检查点

python

1from langgraph.checkpoint.sqlite import SqliteSaver
2
3# 创建持久化存储
4memory = SqliteSaver.from_conn_string("checkpoints.db")
5
6# 编译时指定checkpointer
7app = workflow.compile(checkpointer=memory)
8
9# 使用thread_id跟踪对话
10config = {"configurable": {"thread_id": "user-123"}}
11
12# 第一次调用
13app.invoke({"input": "你好"}, config)
14
15# 稍后可以继续同一对话
16app.invoke({"input": "继续之前的话题"}, config)
17
18# 查看历史状态
19for state in app.get_state_history(config):
20    print(state)

2. 子图(Subgraphs)

python

1# 定义子图
2def create_subworkflow():
3    subgraph = Graph()
4    
5    subgraph.add_node("step1", step1_node)
6    subgraph.add_node("step2", step2_node)
7    subgraph.add_edge("step1", "step2")
8    subgraph.add_edge("step2", END)
9    subgraph.set_entry_point("step1")
10    
11    return subgraph.compile()
12
13# 在主图中使用子图
14main_workflow = Graph()
15
16sub_app = create_subworkflow()
17
18def use_subgraph(state: State) -> State:
19    result = sub_app.invoke({"data": state["data"]})
20    state["result"] = result
21    return state
22
23main_workflow.add_node("process", use_subgraph)

3. 流式输出

python

1# 流式执行
2for chunk in app.stream({"input": "你好"}):
3    print(chunk)
4
5# 异步流式
6async for chunk in app.astream({"input": "你好"}):
7    print(chunk)

4. 并行执行

python

1from langgraph.graph import Graph
2
3# 定义可并行的节点
4workflow.add_node("task1", task1_node)
5workflow.add_node("task2", task2_node)
6workflow.add_node("task3", task3_node)
7workflow.add_node("merge", merge_node)
8
9# 从同一节点发出多条边 - 并行执行
10workflow.add_edge("start", "task1")
11workflow.add_edge("start", "task2")
12workflow.add_edge("start", "task3")
13
14# 汇聚到merge节点
15workflow.add_edge("task1", "merge")
16workflow.add_edge("task2", "merge")
17workflow.add_edge("task3", "merge")

可视化和调试

可视化图结构

python

1from IPython.display import Image, display
2
3# 生成图的Mermaid表示
4graph_diagram = app.get_graph().draw_mermaid()
5print(graph_diagram)
6
7# 在Jupyter中显示
8display(Image(app.get_graph().draw_mermaid_png()))

调试技巧

python

1# 1. 使用verbose模式
2app = workflow.compile(debug=True)
3
4# 2. 检查中间状态
5def debug_node(state: State) -> State:
6    print(f"当前状态: {state}")
7    return state
8
9workflow.add_node("debug", debug_node)
10
11# 3. 使用断点
12def breakpoint_node(state: State) -> State:
13    import pdb; pdb.set_trace()
14    return state

最佳实践

1. 状态设计

python

1# ✅ 好的状态设计
2class GoodState(TypedDict):
3    # 清晰的命名
4    user_input: str
5    llm_response: str
6    
7    # 使用Annotated处理累加
8    conversation_history: Annotated[list, add]
9    
10    # 包含元数据
11    timestamp: str
12    user_id: str
13
14# ❌ 不好的状态设计
15class BadState(TypedDict):
16    data: dict  # 太模糊
17    stuff: list  # 不清楚用途
18    x: str  # 无意义的命名

2. 节点职责单一

python

1# ✅ 职责单一
2def retrieve_node(state):
3    # 只负责检索
4    docs = vectorstore.search(state["query"])
5    return {"documents": docs}
6
7def generate_node(state):
8    # 只负责生成
9    answer = llm.generate(state["documents"])
10    return {"answer": answer}
11
12# ❌ 职责混乱
13def do_everything_node(state):
14    docs = vectorstore.search(state["query"])
15    answer = llm.generate(docs)
16    score = evaluate(answer)
17    # 做太多事情
18    return {"result": ...}

3. 错误处理

python

1def safe_node(state: State) -> State:
2    try:
3        result = process(state["data"])
4        return {"result": result, "error": None}
5    except Exception as e:
6        return {
7            "result": None,
8            "error": str(e),
9            "needs_retry": True
10        }
11
12# 在路由中处理错误
13def error_router(state: State) -> str:
14    if state.get("error"):
15        if state.get("retry_count", 0) < 3:
16            return "retry"
17        return "error_handler"
18    return "continue"

与其他框架对比

特性	LangGraph	LangChain	CrewAI	AutoGen
循环支持	✅ 原生支持	❌ 不支持	✅ 支持	✅ 支持
条件分支	✅ 灵活	⚠️ 有限	⚠️ 有限	✅ 灵活
状态管理	✅ 强大	⚠️ 基础	⚠️ 基础	✅ 强大
可视化	✅ 内置	❌ 无	❌ 无	⚠️ 有限
学习曲线	中等	低	低	高
灵活性	极高	中	中	高

总结

LangGraph是构建复杂Agent系统的强大工具，适合：

✅ 需要循环和条件判断的场景
✅ 多Agent协作系统
✅ 需要人工干预的工作流
✅ 复杂的决策流程
✅ 需要状态持久化的应用

下一步

LangChain基础 - 了解LangChain生态
Agent设计模式 - 学习常见的Agent模式
生产部署 - 部署LangGraph应用

参考资源

forum

评论区 / Comments

share分享文章 / Share

什么是LangGraph​

为什么需要LangGraph​

核心概念​

快速开始​

安装​

Hello World​

核心组件详解​

1. State (状态)​

2. Nodes (节点)​

3. Edges (边)​

实战项目​

项目1: 简单的对话Agent​

项目2: 多Agent协作系统​

项目3: RAG增强的问答系统​

项目4: 人工审批流程​

高级特性​

1. 持久化和检查点​

2. 子图(Subgraphs)​

3. 流式输出​

4. 并行执行​

可视化和调试​

可视化图结构​

调试技巧​

最佳实践​

1. 状态设计​

2. 节点职责单一​

3. 错误处理​

与其他框架对比​

总结​

参考资源​

评论区 / Comments

什么是LangGraph

为什么需要LangGraph

核心概念

快速开始

安装

Hello World

核心组件详解

1. State (状态)

2. Nodes (节点)

3. Edges (边)

实战项目

项目1: 简单的对话Agent

项目2: 多Agent协作系统

项目3: RAG增强的问答系统

项目4: 人工审批流程

高级特性

1. 持久化和检查点

2. 子图(Subgraphs)

3. 流式输出

4. 并行执行

可视化和调试

可视化图结构

调试技巧

最佳实践

1. 状态设计

2. 节点职责单一

3. 错误处理

与其他框架对比

总结

参考资源