本课学习LangGraph——基于图结构的Agent框架。它将Agent流程建模为有向图：节点(Node)是处理步骤，边(Edge)是流转路径，状态(State)在节点间传递。这种方式让Agent流程可视化、可中断、可调试。
第20课: Agent框架(LangGraph)
├── 核心概念
│ ├── State: 节点间传递的共享数据
│ ├── Node: 执行特定任务的函数
│ └── Edge: 节点间连接(可带条件)
├── StateGraph
│ ├── add_node() 注册节点
│ ├── add_edge() 连接边
│ ├── add_conditional_edges() 条件路由
│ └── run() 执行图
├── 条件边与循环
│ ├── 条件边: 根据状态选择路径
│ └── 循环边: agent→think→act→think
└── 人机协作
├── interrupt_before: 节点前中断
├── interrupt_after: 节点后中断
└── human_approval_fn: 审批函数
# LangGraph核心概念: State/Node/Edge
from typing import TypedDict, Dict, List, Optional, Callable
from dataclasses import dataclass, field
@dataclass
class GraphState:
"""图状态: 在节点间传递的共享数据"""
messages: List[dict] = field(default_factory=list)
current_node: str = ""
iteration: int = 0
max_iterations: int = 10
task: str = ""
result: str = ""
error: Optional[str] = None
metadata: Dict = field(default_factory=dict)
def add_message(self, role, content):
self.messages.append({"role": role, "content": content})
def get_last_message(self) -> Optional[dict]:
return self.messages[-1] if self.messages else None
@dataclass
class GraphNode:
"""图节点: 执行特定任务的函数"""
name: str
handler: Callable
description: str = ""
def execute(self, state: GraphState) -> GraphState:
return self.handler(state)
@dataclass
class GraphEdge:
"""图边: 节点间的连接，可带条件"""
source: str
target: str
condition: Optional[Callable] = None # 条件函数: state -> bool
label: str = ""
def should_traverse(self, state: GraphState) -> bool:
if self.condition is None: return True
return self.condition(state)✅ 验证通过：State/Node/Edge三要素均有数据类定义和执行方法
# StateGraph: LangGraph风格的状态图实现
from typing import Callable, Dict, List, Optional, Any
from dataclasses import dataclass, field
import json
@dataclass
class StateGraph:
"""LangGraph风格的状态图 - 节点注册/边连接/条件路由/执行"""
name: str = "StateGraph"
nodes: Dict[str, Callable] = field(default_factory=dict)
edges: List[dict] = field(default_factory=list) # {source, target, condition, label}
entry_point: Optional[str] = None
exit_points: List[str] = field(default_factory=list)
interrupt_before: List[str] = field(default_factory=list)
interrupt_after: List[str] = field(default_factory=list)
def add_node(self, name: str, handler: Callable) -> "StateGraph":
"""注册节点"""
self.nodes[name] = handler
return self
def add_edge(self, source: str, target: str, condition: Callable = None, label: str = "") -> "StateGraph":
"""添加边（可带条件）"""
self.edges.append({"source": source, "target": target, "condition": condition, "label": label})
return self
def add_conditional_edges(self, source: str, condition_fn: Callable, mappings: Dict[str, str]) -> "StateGraph":
"""添加条件边: condition_fn(state) -> key, 映射到对应目标节点"""
for key, target in mappings.items():
def make_cond(k, fn):
return lambda state, _k=k, _fn=fn: _fn(state) == _k
self.add_edge(source, target, condition=make_cond(key, condition_fn), label=key)
return self
def set_entry_point(self, name: str) -> "StateGraph":
self.entry_point = name
return self
def set_finish_point(self, name: str) -> "StateGraph":
self.exit_points.append(name)
return self
def _get_next_node(self, current: str, state: dict) -> Optional[str]:
"""根据当前节点和状态，确定下一个节点"""
candidates = [e for e in self.edges if e["source"] == current]
for edge in candidates:
if edge["condition"] is None or edge["condition"](state):
return edge["target"]
return None
def run(self, initial_state: dict, max_steps: int = 20) -> dict:
"""执行状态图"""
state = dict(initial_state)
state["_current_node"] = self.entry_point
state["_step"] = 0
state["_path"] = []
for _ in range(max_steps):
current = state.get("_current_node")
if not current or current not in self.nodes:
break
state["_step"] += 1
state["_path"].append(current)
# 执行当前节点
handler = self.nodes[current]
state = handler(state)
state["_current_node"] = current # 确保更新
# 检查是否到达终点
if current in self.exit_points:
break
# 获取下一个节点
next_node = self._get_next_node(current, state)
if next_node:
state["_current_node"] = next_node
else:
break
state.pop("_current_node", None)
return state
# 使用示例
def agent_think(state):
state["thought"] = "分析任务需求..."
state["needs_tool"] = True
return state
def agent_act(state):
state["action_result"] = "工具执行完成"
state["task_complete"] = False
return state
def agent_respond(state):
state["result"] = "最终答案"
return state
def route_after_think(state):
return "act" if state.get("needs_tool") else "respond"
graph = StateGraph(name="SimpleAgent")
graph.add_node("think", agent_think)
graph.add_node("act", agent_act)
graph.add_node("respond", agent_respond)
graph.set_entry_point("think")
graph.add_conditional_edges("think", route_after_think, {"act": "act", "respond": "respond"})
graph.add_edge("act", "think") # 循环: act → think
graph.set_finish_point("respond")
result = graph.run({"task": "分析数据"})
print(f"执行路径: {result.get('_path', [])}")
print(f"结果: {result.get('result', '无')}")✅ 验证通过：StateGraph实现了节点注册、边连接、条件路由和完整执行流程
条件边根据状态动态选择路径，循环边实现Agent的反复推理-行动循环。
条件边示例:
think → [needs_tool=True] → act
think → [needs_tool=False] → respond
循环边示例:
act → think → act → think → respond
完整流程:
START → think ⇄ act → respond → END
↑ |
└──────┘
# LangGraph人机协作: interrupt模式
from typing import Callable, Dict, Optional
from dataclasses import dataclass, field
@dataclass
class HumanInTheLoopGraph:
"""带人机协作的状态图 - interrupt/审批/渐进自主"""
name: str = "HITLGraph"
nodes: Dict[str, Callable] = field(default_factory=dict)
edges: list = field(default_factory=list)
entry_point: Optional[str] = None
exit_points: list = field(default_factory=list)
interrupt_before: list = field(default_factory=list)
interrupt_after: list = field(default_factory=list)
_paused_at: Optional[str] = None
_paused_state: Optional[dict] = None
def add_node(self, name, handler): self.nodes[name] = handler; return self
def add_edge(self, source, target, condition=None): self.edges.append({"source": source, "target": target, "condition": condition}); return self
def set_entry_point(self, name): self.entry_point = name; return self
def set_finish_point(self, name): self.exit_points.append(name); return self
def add_interrupt_before(self, node): self.interrupt_before.append(node); return self
def add_interrupt_after(self, node): self.interrupt_after.append(node); return self
def _get_next(self, current, state):
for e in self.edges:
if e["source"] == current and (e["condition"] is None or e["condition"](state)):
return e["target"]
return None
def run(self, initial_state, max_steps=20, human_approval_fn=None):
"""执行状态图，在interrupt节点暂停等待人工确认"""
state = dict(initial_state)
state["_current"] = self.entry_point
state["_path"] = []
for _ in range(max_steps):
current = state.get("_current")
if not current or current not in self.nodes: break
state["_path"].append(current)
# interrupt_before检查
if current in self.interrupt_before:
if human_approval_fn and not human_approval_fn(current, state):
state["result"] = f"在 {current} 被人工拒绝"
return state
# 执行节点
state = self.nodes[current](state)
state["_current"] = current
# interrupt_after检查
if current in self.interrupt_after:
if human_approval_fn and not human_approval_fn(current, state):
state["result"] = f"在 {current} 后被人工拒绝"
return state
if current in self.exit_points: break
next_node = self._get_next(current, state)
if next_node: state["_current"] = next_node
else: break
return state
# 使用: 带审批的Agent
def plan(state):
state["plan"] = ["步骤1: 收集数据", "步骤2: 分析数据", "步骤3: 生成报告"]
return state
def execute(state):
state["execution_result"] = "数据收集完成，分析完成"
return state
def finalize(state):
state["result"] = "最终报告已生成"
return state
def approve_plan(node, state):
"""审批函数: 模拟人工审批"""
print(f"[审批] 节点: {node}, 计划: {state.get('plan', [])}")
return True # True=批准, False=拒绝
hitl = HumanInTheLoopGraph(name="ApprovalAgent")
hitl.add_node("plan", plan)
hitl.add_node("execute", execute)
hitl.add_node("finalize", finalize)
hitl.set_entry_point("plan")
hitl.add_edge("plan", "execute")
hitl.add_edge("execute", "finalize")
hitl.set_finish_point("finalize")
hitl.add_interrupt_before("execute") # 执行前需审批
result = hitl.run({"task": "生成季度报告"}, human_approval_fn=approve_plan)
print(f"路径: {result.get('_path', [])}")
print(f"结果: {result.get('result', '无')}")✅ 验证通过：HITL图实现了interrupt_before/after和human_approval_fn审批机制
| 特性 | 普通Agent | LangGraph |
|---|---|---|
| 流程可视化 | ❌ 不可见 | ✅ 图结构清晰 |
| 中断恢复 | ❌ 不支持 | ✅ interrupt机制 |
| 条件路由 | ❌ 硬编码 | ✅ conditional_edges |
| 调试追踪 | 困难 | ✅ path追踪 |
# 挑战: 构建一个RAG Agent图
# 节点: retrieve → grade → generate
# 条件边: grade决定重试还是生成
# interrupt: 生成前人工确认实现图嵌套——子图作为节点嵌入主图