9470 字
约 31 分钟
1
第15章 LangGraph与状态机
2026-09-18
第15章 LangGraph与状态机
来源：https://ai-agent-guide.xiaofuge.cn/chapters/ch11-langgraph.html 所属：第四篇-协作与编排
用图结构构建可控的 Agent 工作流
15.1 LangGraph：把 Agent 流程变成图
LangChain 团队推出的 LangGraph 是 2024 年最火的 Agent 框架之一。核心思想：把 Agent 工作流建模为有向图。
Agent 的每一步是图中的一个节点（Node），步骤之间的跳转是边（Edge）。整个 Agent 就是一张图，可视化、可调试、可控。
15.2 LangGraph 的核心概念
📐 五个核心概念
1. State（状态）
全局状态对象，所有节点共享。用 TypedDict 定义，包含消息历史、中间结果等。每个节点可以读写 State。
2. Node（节点）
图中的执行单元。每个节点是一个函数，接收 State，返回更新后的 State。节点 = 一步操作。
3. Edge（边）
节点之间的连接。普通边 = 固定跳转；条件边 = 根据 State 动态决定下一个节点。
4. Conditional Edge（条件边）
根据当前 State 动态路由。类似 if-else：if state['need_search'] → search_node, else → code_node。
5. Checkpoint（检查点）
每个节点执行后自动保存 State 快照。支持时间旅行（回到任意检查点）、人工干预（暂停在某节点等人审核后继续）、错误恢复（从失败点重试）。
15.3 用 LangGraph 构建 ReAct Agent
from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated
from langchain_core.messages import HumanMessage, AIMessage
import operator
# 1. 定义状态
class AgentState(TypedDict):
messages: Annotated[list, operator.add] # 消息列表（只增不覆盖）
tool_results: list # 工具结果
# 2. 定义节点函数
def call_llm(state: AgentState):
"""LLM 推理节点：决定下一步做什么"""
response = llm.invoke(state["messages"])
return {"messages": [response]}
def call_tool(state: AgentState):
"""工具执行节点：执行 LLM 选择的工具"""
last_msg = state["messages"][-1]
tool_call = last_msg.tool_calls[0]
result = execute_tool(tool_call)
return {"messages": [ToolMessage(result)]}
# 3. 定义条件路由
def should_continue(state: AgentState):
"""条件边：判断是否继续循环"""
last_msg = state["messages"][-1]
if last_msg.tool_calls:
return "tools" # 需要调用工具
return END # 不需要，结束
# 4. 构建图
workflow = StateGraph(AgentState)
# 添加节点
workflow.add_node("agent", call_llm)
workflow.add_node("tools", call_tool)
# 设置入口
workflow.set_entry_point("agent")
# 添加边和条件边
workflow.add_conditional_edges(
"agent", # 源节点
should_continue, # 路由函数
{
"tools": "tools", # 返回 "tools" → 跳到 tools 节点
END: END # 返回 END → 结束
}
)
workflow.add_edge("tools", "agent") # tools 执行完回到 agent
# 5. 编译并运行
app = workflow.compile(checkpointer=MemorySaver())
result = app.invoke(
{"messages": [HumanMessage("北京明天天气？")]},
config={"configurable": {"thread_id": "1"}}
)
import { StateGraph, END, Annotation } from "@langchain/langgraph";
import { MemorySaver } from "@langchain/langgraph";
import { HumanMessage, AIMessage, ToolMessage } from "@langchain/core/messages";
// 1. 定义状态（使用 Annotation 声明合并策略）
const AgentState = Annotation.Root({
messages: Annotation({
reducer: (prev, next) => [...prev, ...next],
default: () => [],
}),
tool_results: Annotation({
reducer: (_, next) => next,
default: () => [],
}),
});
// 2. 定义节点函数
function callLLM(state: typeof AgentState.State) {
/** LLM 推理节点：决定下一步做什么 */
const response = llm.invoke(state.messages);
return { messages: [response] };
}
function callTool(state: typeof AgentState.State) {
/** 工具执行节点：执行 LLM 选择的工具 */
const lastMsg = state.messages[state.messages.length - 1] as AIMessage;
const toolCall = lastMsg.tool_calls![0];
const result = executeTool(toolCall);
return { messages: [new ToolMessage(result)] };
}
// 3. 定义条件路由
function shouldContinue(state: typeof AgentState.State): string {
/** 条件边：判断是否继续循环 */
const lastMsg = state.messages[state.messages.length - 1] as AIMessage;
if (lastMsg.tool_calls && lastMsg.tool_calls.length > 0) {
return "tools"; // 需要调用工具
}
return END; // 不需要，结束
}
// 4. 构建图
const workflow = new StateGraph(AgentState);
// 添加节点
workflow.addNode("agent", callLLM);
workflow.addNode("tools", callTool);
// 设置入口
workflow.setEntryPoint("agent");
// 添加边和条件边
workflow.addConditionalEdges("agent", shouldContinue, {
tools: "tools",
[END]: END,
});
workflow.addEdge("tools", "agent");
// 5. 编译并运行
const app = workflow.compile({ checkpointer: new MemorySaver() });
const result = await app.invoke(
{ messages: [new HumanMessage("北京明天天气？")] },
{ configurable: { thread_id: "1" } }
);
package main
import (
"fmt"
)
// Message 表示对话中的一条消息
type Message struct {
Role string `json:"role"`
Content string `json:"content"`
ToolCalls []ToolCall `json:"tool_calls,omitempty"`
}
// ToolCall 表示 LLM 返回的工具调用请求
type ToolCall struct {
Name string `json:"name"`
Args map[string]any `json:"args"`
}
// AgentState 全局状态，所有节点共享
type AgentState struct {
Messages []Message `json:"messages"`
ToolResults []any `json:"tool_results"`
}
// NodeFunc 节点函数类型：接收当前状态，返回状态更新
type NodeFunc func(AgentState) map[string]any
// RouteFunc 条件路由函数类型：返回下一个节点名
type RouteFunc func(AgentState) string
// 1. 定义节点函数
func callLLM(state AgentState) map[string]any {
// LLM 推理节点：决定下一步做什么
// 实际项目中这里调用 LLM API
response := Message{Role: "ai", Content: "", ToolCalls: []ToolCall{{Name: "get_weather", Args: map[string]any{"city": "北京"}}}}
return map[string]any{"messages": append(state.Messages, response)}
}
func callTool(state AgentState) map[string]any {
// 工具执行节点：执行 LLM 选择的工具
lastMsg := state.Messages[len(state.Messages)-1]
if len(lastMsg.ToolCalls) == 0 {
return map[string]any{"messages": state.Messages}
}
toolCall := lastMsg.ToolCalls[0]
result := executeTool(toolCall) // 实际工具执行
toolMsg := Message{Role: "tool", Content: result}
return map[string]any{"messages": append(state.Messages, toolMsg)}
}
func executeTool(tc ToolCall) string {
// 实际项目中根据 tc.Name 调用对应工具
return fmt.Sprintf("{\"city\": \"%s\", \"weather\": \"晴\"}", tc.Args["city"])
}
// 2. 定义条件路由
func shouldContinue(state AgentState) string {
// 条件边：判断是否继续循环
lastMsg := state.Messages[len(state.Messages)-1]
if len(lastMsg.ToolCalls) > 0 {
return "tools" // 需要调用工具
}
return "end" // 不需要，结束
}
// 3. 图引擎：用状态机模拟 StateGraph
type StateGraph struct {
nodes map[string]NodeFunc
edges map[string]string // 普通边: from -> to
condEdges map[string]RouteFunc // 条件边: from -> router
routeMap map[string]map[string]string // 条件边路由表
entryPoint string
}
func NewStateGraph() *StateGraph {
return &StateGraph{
nodes: make(map[string]NodeFunc),
edges: make(map[string]string),
condEdges: make(map[string]RouteFunc),
routeMap: make(map[string]map[string]string),
}
}
func (g *StateGraph) AddNode(name string, fn NodeFunc) { g.nodes[name] = fn }
func (g *StateGraph) SetEntryPoint(name string) { g.entryPoint = name }
func (g *StateGraph) AddEdge(from, to string) { g.edges[from] = to }
func (g *StateGraph) AddConditionalEdges(from string, router RouteFunc, routes map[string]string) {
g.condEdges[from] = router
g.routeMap[from] = routes
}
// 4. 构建图
func buildGraph() *StateGraph {
workflow := NewStateGraph()
workflow.AddNode("agent", callLLM)
workflow.AddNode("tools", callTool)
workflow.SetEntryPoint("agent")
workflow.AddConditionalEdges("agent", shouldContinue, map[string]string{
"tools": "tools",
"end": "end",
})
workflow.AddEdge("tools", "agent")
return workflow
}
// 5. 编译并运行（模拟 invoke）
func (g *StateGraph) Invoke(initial AgentState) AgentState {
state := initial
current := g.entryPoint
for step := 0; step messages = new ArrayList<>();
List toolResults = new ArrayList<>();
AgentState copy() {
AgentState s = new AgentState();
s.messages = new ArrayList<>(this.messages);
s.toolResults = new ArrayList<>(this.toolResults);
return s;
}
}
static class Message {
String role;
String content;
List toolCalls;
Message(String role, String content) { this.role = role; this.content = content; }
Message(String role, String content, List toolCalls) {
this.role = role; this.content = content; this.toolCalls = toolCalls;
}
}
static class ToolCall {
String name;
Map args;
ToolCall(String name, Map args) { this.name = name; this.args = args; }
}
// ── 2. 定义节点函数 ──
static Map callLLM(AgentState state) {
// LLM 推理节点：决定下一步做什么
// 实际项目中调用 llm.invoke(state.messages)
Message response = new Message("ai", "",
List.of(new ToolCall("get_weather", Map.of("city", "北京"))));
List newMsgs = new ArrayList<>(state.messages);
newMsgs.add(response);
return Map.of("messages", newMsgs);
}
static Map callTool(AgentState state) {
// 工具执行节点：执行 LLM 选择的工具
Message lastMsg = state.messages.get(state.messages.size() - 1);
if (lastMsg.toolCalls == null || lastMsg.toolCalls.isEmpty()) {
return Map.of("messages", state.messages);
}
ToolCall toolCall = lastMsg.toolCalls.get(0);
String result = executeTool(toolCall);
List newMsgs = new ArrayList<>(state.messages);
newMsgs.add(new Message("tool", result));
return Map.of("messages", newMsgs);
}
static String executeTool(ToolCall tc) {
// 实际项目中根据 tc.name 调用对应工具
return "{\"city\": \"" + tc.args.get("city") + "\", \"weather\": \"晴\"}";
}
// ── 3. 定义条件路由 ──
static String shouldContinue(AgentState state) {
Message lastMsg = state.messages.get(state.messages.size() - 1);
if (lastMsg.toolCalls != null && !lastMsg.toolCalls.isEmpty()) {
return "tools"; // 需要调用工具
}
return "end"; // 不需要，结束
}
// ── 4. 图引擎：状态机模拟 StateGraph ──
interface NodeFunc { Map apply(AgentState state); }
interface RouteFunc { String apply(AgentState state); }
static class StateGraph {
Map nodes = new LinkedHashMap<>();
Map edges = new HashMap<>();
Map condEdges = new HashMap<>();
Map> routeMap = new HashMap<>();
String entryPoint;
void addNode(String name, NodeFunc fn) { nodes.put(name, fn); }
void setEntryPoint(String name) { entryPoint = name; }
void addEdge(String from, String to) { edges.put(from, to); }
void addConditionalEdges(String from, RouteFunc router, Map routes) {
condEdges.put(from, router);
routeMap.put(from, routes);
}
AgentState invoke(AgentState initial) {
AgentState state = initial;
String current = entryPoint;
for (int step = 0; step updates = fn.apply(state);
Object msgs = updates.get("messages");
if (msgs instanceof List) {
@SuppressWarnings("unchecked")
List newMsgs = (List) msgs;
state = new AgentState();
state.messages = new ArrayList<>(newMsgs);
state.toolResults = new ArrayList<>();
}
// 路由判断
RouteFunc router = condEdges.get(current);
if (router != null) {
String nextKey = router.apply(state);
Map routes = routeMap.get(current);
if (routes != null && routes.containsKey(nextKey)) {
String nextNode = routes.get(nextKey);
if ("end".equals(nextNode)) break;
current = nextNode;
continue;
}
}
String next = edges.get(current);
if (next != null) { current = next; } else { break; }
}
return state;
}
}
// ── 5. 构建图并运行 ──
public static void main(String[] args) {
StateGraph workflow = new StateGraph();
workflow.addNode("agent", ReActAgent::callLLM);
workflow.addNode("tools", ReActAgent::callTool);
workflow.setEntryPoint("agent");
workflow.addConditionalEdges("agent", ReActAgent::shouldContinue, Map.of(
"tools", "tools", "end", "end"));
workflow.addEdge("tools", "agent");
AgentState initial = new AgentState();
initial.messages.add(new Message("user", "北京明天天气？"));
AgentState result = workflow.invoke(initial);
System.out.println("最终消息数: " + result.messages.size());
System.out.println("最后一条: " + result.messages.get(result.messages.size() - 1).content);
}
}
代码解析
State 定义
Annotated[list, operator.add] 表示 messages 列表用"追加"而非"覆盖"语义。新消息追加到列表，不替换旧消息。
条件边
should_continue 函数检查最后一条消息是否包含 tool_calls。有则路由到 tools 节点，无则结束。
循环结构
agent → (条件) → tools → agent → ... 形成循环。这就是 ReAct 的循环在 LangGraph 中的表达。
Checkpointer
MemorySaver() 保存每步状态。通过 thread_id 可以回到任意检查点，支持人工干预。
15.4 Human-in-the-Loop：人工干预
LangGraph 的 Checkpoint 机制让"人工干预"变得自然：
人工干预代码示例
# 在工具执行前暂停，等待人工审核
app = workflow.compile(
checkpointer=MemorySaver(),
interrupt_before=["tools"] # 在 tools 节点前暂停
)
# 第一次调用 → 执行到 tools 前暂停
result = app.invoke(
{"messages": [HumanMessage("删除 /tmp 目录")]},
config={"configurable": {"thread_id": "1"}}
)
# 此时 Agent 想调用 rm 命令，暂停等待人工审核
# 人工审核后，决定继续或修改
# 方式1: 直接继续（批准）
result = app.invoke(None, config={"configurable": {"thread_id": "1"}})
# 方式2: 修改 State 后继续（纠正）
app.update_state(
config={"configurable": {"thread_id": "1"}},
values={"messages": [HumanMessage("不要删除，改为列出文件")]}
)
result = app.invoke(None, config={"configurable": {"thread_id": "1"}})
15.5 LangGraph vs LangChain Agent | 对比项 | LangChain Agent | LangGraph | | --- | --- | --- | | 流程控制 | 黑盒，LLM 自主决定 | 白盒，图结构显式定义 | | 可调试性 | 难，只能看最终输出 | 强，每步可检查 State | | 人工干预 | 不支持 | 原生支持（Checkpoint） | | 并行执行 | 不支持 | 支持（多个节点并行） | | 适用场景 | 简单、快速的 Agent | 复杂、需要控制的生产级 Agent | ## 15.6 LangGraph 完整实战代码
前面几节我们分别学习了 State、Node、Edge 和 Checkpoint 的概念，现在把它们组合起来，写一个完整可运行的 LangGraph 项目。这个示例模拟一个智能客服 Agent：接收用户问题，判断是否需要查数据库、调用外部 API 或直接回答，执行后在审查节点检查结果质量，不达标则重试，达标则返回最终回复。
from langgraph.graph import StateGraph, END, START
from langgraph.checkpoint.memory import MemorySaver
from langgraph.prebuilt import ToolNode
from typing import TypedDict, Annotated
import operator
# ── 1. 定义全局状态 ──
class AgentState(TypedDict):
messages: Annotated[list, operator.add] # 对话历史（追加策略）
step_count: int # 当前循环步数
tool_results: list # 工具返回结果
needs_review: bool # 是否需要人工审查
# ── 2. 定义节点函数 ──
def agent_node(state: AgentState) -> dict:
"""LLM 推理节点：分析用户意图，决定调用哪个工具"""
step = state.get("step_count", 0) + 1
# 实际项目中这里调用 llm.invoke(state["messages"])
decision = "search_db" if step dict:
"""数据库查询节点：根据 LLM 决策检索知识库"""
# 实际项目中这里连接数据库执行查询
result = {"found": True, "data": "订单 #12345 已发货，预计 7月3日 送达"}
return {"tool_results": [result], "messages": [{"role": "tool", "content": str(result)}]}
def answer_node(state: AgentState) -> dict:
"""最终回复节点：整合所有信息生成回答"""
final = f"根据查询结果：{state.get('tool_results', [])}"
return {"messages": [{"role": "ai", "content": final}]}
def review_node(state: AgentState) -> dict:
"""质量审查节点：检查回复是否满足要求"""
last = state["messages"][-1]
quality_ok = len(str(last.get("content", ""))) > 5
return {"needs_review": not quality_ok, "step_count": state.get("step_count", 0)}
# ── 3. 定义条件路由 ──
def should_continue(state: AgentState) -> str:
if state.get("step_count", 0) >= 5:
return "end"
last_msg = state["messages"][-1]
if "decision=search_db" in str(last_msg.get("content", "")):
return "search_db"
return "answer"
def review_decision(state: AgentState) -> str:
if state.get("needs_review", False) and state.get("step_count", 0) dict:
/** docstring */
// step = state.get("step_count", 0) + 1
// 实际项目中这里调用 llm.invoke(state["messages"])
// decision = "search_db" if step dict:
/** docstring */
// 实际项目中这里连接数据库执行查询
// result = {"found": True, "data": "订单 #12345 已发货，预计 7月3日 送达"}
return {"tool_results": [result], "messages": [{"role": "tool", "content": str(result)}]};
// def answer_node(state: AgentState) -> dict:
/** docstring */
// final = f"根据查询结果：{state.get('tool_results', [])}"
return {"messages": [{"role": "ai", "content": final}]};
// def review_node(state: AgentState) -> dict:
/** docstring */
// last = state["messages"][-1]
// quality_ok = len(str(last.get("content", ""))) > 5
return {"needs_review": not quality_ok, "step_count": state.get("step_count", 0)};
// ── 3. 定义条件路由 ──
// def should_continue(state: AgentState) -> str:
if (state.get("step_count", 0) >= 5) {
return "end";
// last_msg = state["messages"][-1]
if ("decision=search_db" in str(last_msg.get("content", ""))) {
return "search_db";
return "answer";
// def review_decision(state: AgentState) -> str:
if (state.get("needs_review", false) && state.get("step_count", 0) dict:
// Python: step = state.get("step_count", 0) + 1
// 实际项目中这里调用 llm.invoke(state["messages"])
// Python: decision = "search_db" if step dict:
// 实际项目中这里连接数据库执行查询
// Python: result = {"found": True, "data": "订单 #12345 已发货，预计 7月3日 送达"}
return {"tool_results": [result], "messages": [{"role": "tool", "content": str(result)}]}
// Python: def answer_node(state: AgentState) -> dict:
// Python: final = f"根据查询结果：{state.get('tool_results', [])}"
return {"messages": [{"role": "ai", "content": final}]}
// Python: def review_node(state: AgentState) -> dict:
// Python: last = state["messages"][-1]
// Python: quality_ok = len(str(last.get("content", ""))) > 5
return {"needs_review": not quality_ok, "step_count": state.get("step_count", 0)}
// ── 3. 定义条件路由 ──
// Python: def should_continue(state: AgentState) -> str:
if state.get("step_count", 0) >= 5 {
return "end"
// Python: last_msg = state["messages"][-1]
if "decision=search_db" in str(last_msg.get("content", "")) {
return "search_db"
return "answer"
// Python: def review_decision(state: AgentState) -> str:
if state.get("needs_review", false) and state.get("step_count", 0) dict:
// Python: step = state.get("step_count", 0) + 1
// 实际项目中这里调用 llm.invoke(state["messages"])
// Python: decision = "search_db" if step dict:
// 实际项目中这里连接数据库执行查询
// Python: result = {"found": True, "data": "订单 #12345 已发货，预计 7月3日 送达"}
return {"tool_results": [result], "messages": [{"role": "tool", "content": str(result)}]};
// Python: def answer_node(state: AgentState) -> dict:
// Python: final = f"根据查询结果：{state.get('tool_results', [])}"
return {"messages": [{"role": "ai", "content": final}]};
// Python: def review_node(state: AgentState) -> dict:
// Python: last = state["messages"][-1]
// Python: quality_ok = len(str(last.get("content", ""))) > 5
return {"needs_review": not quality_ok, "step_count": state.get("step_count", 0)};
// ── 3. 定义条件路由 ──
// Python: def should_continue(state: AgentState) -> str:
if (state.get("step_count", 0) >= 5) {
return "end";
// Python: last_msg = state["messages"][-1]
if ("decision=search_db" in str(last_msg.get("content", ""))) {
return "search_db";
return "answer";
// Python: def review_decision(state: AgentState) -> str:
if (state.get("needs_review", false) && state.get("step_count", 0) dict:
"""根据用户问题生成多个搜索查询"""
return {"queries": ["LangGraph 教程", "LangGraph 实战", "LangGraph vs LangChain"]}
def execute_search(state: ResearchState) -> dict:
"""执行搜索并收集结果"""
results = []
for q in state["queries"]:
# 实际项目中调用搜索 API
results.append({"query": q, "snippet": f"关于 {q} 的搜索结果..."})
return {"findings": results}
def summarize(state: ResearchState) -> dict:
"""汇总搜索结果"""
summary = "; ".join([r["snippet"] for r in state["findings"]])
return {"findings": [{"summary": summary}]}
# ── 构建子图 ──
research_subgraph = StateGraph(ResearchState)
research_subgraph.add_node("gen_queries", generate_queries)
research_subgraph.add_node("search", execute_search)
research_subgraph.add_node("summarize", summarize)
research_subgraph.add_edge(START, "gen_queries")
research_subgraph.add_edge("gen_queries", "search")
research_subgraph.add_edge("search", "summarize")
research_subgraph.add_edge("summarize", END)
research_app = research_subgraph.compile()
# ── 主图：将子图作为节点嵌入 ──
class MainState(TypedDict):
topic: str
research_result: str
final_report: str
def research_node(state: MainState) -> dict:
"""调用研究子图"""
sub_result = research_app.invoke({"queries": [], "findings": []})
return {"research_result": str(sub_result["findings"][-1])}
def write_report(state: MainState) -> dict:
"""根据研究结果撰写报告"""
return {"final_report": f"研究报告：{state['research_result']}"}
main_graph = StateGraph(MainState)
main_graph.add_node("research", research_node)
main_graph.add_node("report", write_report)
main_graph.add_edge(START, "research")
main_graph.add_edge("research", "report")
main_graph.add_edge("report", END)
main_app = main_graph.compile()
import {StateGraph, END, START} from 'langgraph.graph';
// TypeScript has built-in types, no import needed for TypedDict, Annotated
import * as operator from 'operator';
// ── 子图状态：研究子流程 ──
class ResearchState {
// queries: list # 查询列表
// findings: Annotated[list, operator.add] # 研究结果
// ── 子图节点 ──
// def generate_queries(state: ResearchState) -> dict:
/** docstring */
return {"queries": ["LangGraph 教程", "LangGraph 实战", "LangGraph vs LangChain"]};
// def execute_search(state: ResearchState) -> dict:
/** docstring */
// results = []
for (const q of state["queries"]) {
// 实际项目中调用搜索 API
// results.append({"query": q, "snippet": f"关于 {q} 的搜索结果..."})
return {"findings": results};
// def summarize(state: ResearchState) -> dict:
/** docstring */
// summary = "; ".join([r["snippet"] for r in state["findings"]])
return {"findings": [{"summary": summary}]};
// ── 构建子图 ──
// research_subgraph = StateGraph(ResearchState)
// research_subgraph.add_node("gen_queries", generate_queries)
// research_subgraph.add_node("search", execute_search)
// research_subgraph.add_node("summarize", summarize)
// research_subgraph.add_edge(START, "gen_queries")
// research_subgraph.add_edge("gen_queries", "search")
// research_subgraph.add_edge("search", "summarize")
// research_subgraph.add_edge("summarize", END)
// research_app = research_subgraph.compile()
// ── 主图：将子图作为节点嵌入 ──
class MainState {
// topic: str
// research_result: str
// final_report: str
// def research_node(state: MainState) -> dict:
/** docstring */
// sub_result = research_app.invoke({"queries": [], "findings": []})
return {"research_result": str(sub_result["findings"][-1])};
// def write_report(state: MainState) -> dict:
/** docstring */
return {"final_report": `研究报告：{state['research_result']}`};
// main_graph = StateGraph(MainState)
// main_graph.add_node("research", research_node)
// main_graph.add_node("report", write_report)
// main_graph.add_edge(START, "research")
// main_graph.add_edge("research", "report")
// main_graph.add_edge("report", END)
// main_app = main_graph.compile()
}
package main
import (
"fmt"
"os"
"os/exec"
"strings"
)
// from langgraph.graph import StateGraph, END, START
// from typing import TypedDict, Annotated
// import operator
// ── 子图状态：研究子流程 ──
// ResearchState - CLI Agent class
type ResearchState struct {
// Python: queries: list # 查询列表
// Python: findings: Annotated[list, operator.add] # 研究结果
// ── 子图节点 ──
// Python: def generate_queries(state: ResearchState) -> dict:
return {"queries": ["LangGraph 教程", "LangGraph 实战", "LangGraph vs LangChain"]}
// Python: def execute_search(state: ResearchState) -> dict:
// Python: results = []
for _, q := range state["queries"] {
// 实际项目中调用搜索 API
// Python: results.append({"query": q, "snippet": f"关于 {q} 的搜索结果..."})
return {"findings": results}
// Python: def summarize(state: ResearchState) -> dict:
// Python: summary = "; ".join([r["snippet"] for r in state["findings"]])
return {"findings": [{"summary": summary}]}
// ── 构建子图 ──
// Python: research_subgraph = StateGraph(ResearchState)
// Python: research_subgraph.add_node("gen_queries", generate_queries)
// Python: research_subgraph.add_node("search", execute_search)
// Python: research_subgraph.add_node("summarize", summarize)
// Python: research_subgraph.add_edge(START, "gen_queries")
// Python: research_subgraph.add_edge("gen_queries", "search")
// Python: research_subgraph.add_edge("search", "summarize")
// Python: research_subgraph.add_edge("summarize", END)
// Python: research_app = research_subgraph.compile()
// ── 主图：将子图作为节点嵌入 ──
// MainState - CLI Agent class
type MainState struct {
// Python: topic: str
// Python: research_result: str
// Python: final_report: str
// Python: def research_node(state: MainState) -> dict:
// Python: sub_result = research_app.invoke({"queries": [], "findings": []})
return {"research_result": str(sub_result["findings"][-1])}
// Python: def write_report(state: MainState) -> dict:
return {"final_report": f"研究报告：{state['research_result']}"}
// Python: main_graph = StateGraph(MainState)
// Python: main_graph.add_node("research", research_node)
// Python: main_graph.add_node("report", write_report)
// Python: main_graph.add_edge(START, "research")
// Python: main_graph.add_edge("research", "report")
// Python: main_graph.add_edge("report", END)
// Python: main_app = main_graph.compile()
}
import java.util.*;
import java.util.concurrent.*;
import java.util.regex.*;
import java.io.*;
// from langgraph.graph import StateGraph, END, START
// from typing import TypedDict, Annotated
// import operator
// ── 子图状态：研究子流程 ──
public class ResearchState {
// Python: queries: list # 查询列表
// Python: findings: Annotated[list, operator.add] # 研究结果
// ── 子图节点 ──
// Python: def generate_queries(state: ResearchState) -> dict:
return {"queries": ["LangGraph 教程", "LangGraph 实战", "LangGraph vs LangChain"]};
// Python: def execute_search(state: ResearchState) -> dict:
// Python: results = []
for (var q : state["queries"]) {
// 实际项目中调用搜索 API
// Python: results.append({"query": q, "snippet": f"关于 {q} 的搜索结果..."})
return {"findings": results};
// Python: def summarize(state: ResearchState) -> dict:
// Python: summary = "; ".join([r["snippet"] for r in state["findings"]])
return {"findings": [{"summary": summary}]};
// ── 构建子图 ──
// Python: research_subgraph = StateGraph(ResearchState)
// Python: research_subgraph.add_node("gen_queries", generate_queries)
// Python: research_subgraph.add_node("search", execute_search)
// Python: research_subgraph.add_node("summarize", summarize)
// Python: research_subgraph.add_edge(START, "gen_queries")
// Python: research_subgraph.add_edge("gen_queries", "search")
// Python: research_subgraph.add_edge("search", "summarize")
// Python: research_subgraph.add_edge("summarize", END)
// Python: research_app = research_subgraph.compile()
// ── 主图：将子图作为节点嵌入 ──
public class MainState {
// Python: topic: str
// Python: research_result: str
// Python: final_report: str
// Python: def research_node(state: MainState) -> dict:
// Python: sub_result = research_app.invoke({"queries": [], "findings": []})
return {"research_result": str(sub_result["findings"][-1])};
// Python: def write_report(state: MainState) -> dict:
return {"final_report": f"研究报告：{state['research_result']}"};
// Python: main_graph = StateGraph(MainState)
// Python: main_graph.add_node("research", research_node)
// Python: main_graph.add_node("report", write_report)
// Python: main_graph.add_edge(START, "research")
// Python: main_graph.add_edge("research", "report")
// Python: main_graph.add_edge("report", END)
// Python: main_app = main_graph.compile()
}
}
这段代码展示了两层图结构：内层 research_subgraph 负责「生成查询 → 执行搜索 → 汇总结果」三步，外层主图调用子图后将结果传给 write_report 节点生成最终报告。子图的 State（ResearchState）与主图的 State（MainState）完全独立，通过 research_node 函数做桥接转换，这种设计让两层的逻辑互不污染。
Human-in-the-Loop：三种人机协同模式
LangGraph 的人机协同不止「暂停-审核-继续」一种模式，实际项目中有三种常见模式：审批模式（执行前等人批准）、纠正模式（暂停后人工修改 State 再继续）、对话模式（Agent 遇到不确定时主动向人提问）。下面用代码演示这三种模式的实现。
from langgraph.graph import StateGraph, END, START
from langgraph.checkpoint.memory import MemorySaver
from typing import TypedDict
class HitlState(TypedDict):
action: str
approved: bool
feedback: str
result: str
# ── 模式1：审批模式 — 执行前等待人工批准 ──
def risky_action(state: HitlState) -> dict:
"""高风险操作节点（如删除文件、发送邮件）"""
return {"result": f"已执行操作: {state['action']}"}
def execute_approval_mode():
"""编译时设置 interrupt_before，在 risky_action 前暂停"""
wg = StateGraph(HitlState)
wg.add_node("risky", risky_action)
wg.add_edge(START, "risky")
wg.add_edge("risky", END)
app = wg.compile(
checkpointer=MemorySaver(),
interrupt_before=["risky"] # 关键：执行前暂停
)
return app
# ── 模式2：纠正模式 — 人工修改 State 后继续 ──
def execute_correction_mode(app, thread_id: str):
"""暂停后人工修改 State，再继续执行"""
cfg = {"configurable": {"thread_id": thread_id}}
# 第一次调用：执行到暂停点
state = app.invoke({"action": "删除 /tmp/all", "approved": False, "feedback": "", "result": ""}, config=cfg)
# 人工审查后修改 State
app.update_state(cfg, values={"action": "列出 /tmp/all 文件而非删除"})
# 继续执行
result = app.invoke(None, config=cfg)
return result
# ── 模式3：对话模式 — Agent 主动向人提问 ──
def ask_human(state: HitlState) -> dict:
"""Agent 遇到不确定时，生成问题等待人类回答"""
question = "我不确定你想删除哪个目录，请指定："
# 实际项目中通过 UI 或 chat 将问题发给用户
return {"result": question, "feedback": "等待用户回答"}
def execute_dialog_mode():
"""Agent 在条件边中判断不确定时跳转到 ask_human 节点"""
wg = StateGraph(HitlState)
wg.add_node("risky", risky_action)
wg.add_node("ask", ask_human)
wg.add_edge(START, "risky")
# 条件边：不确定时跳到 ask 节点
wg.add_conditional_edges("risky", lambda s: "ask" if not s.get("approved") else END,
{"ask": "ask", END: END})
wg.add_edge("ask", END)
return wg.compile(checkpointer=MemorySaver())
import {StateGraph, END, START} from 'langgraph.graph';
import {MemorySaver} from 'langgraph.checkpoint.memory';
// TypeScript has built-in types, no import needed for TypedDict
class HitlState {
// action: str
// approved: bool
// feedback: str
// result: str
// ── 模式1：审批模式 — 执行前等待人工批准 ──
// def risky_action(state: HitlState) -> dict:
/** docstring */
return {"result": `已执行操作: {state['action']}`};
function execute_approval_mode() {
/** docstring */
// wg = StateGraph(HitlState)
// wg.add_node("risky", risky_action)
// wg.add_edge(START, "risky")
// wg.add_edge("risky", END)
// app = wg.compile(
// checkpointer = MemorySaver(),
// interrupt_before = ["risky"] # 关键：执行前暂停
// )
return app;
// ── 模式2：纠正模式 — 人工修改 State 后继续 ──
function execute_correction_mode(app, thread_id: string) {
/** docstring */
// cfg = {"configurable": {"thread_id": thread_id}}
// 第一次调用：执行到暂停点
// state = app.invoke({"action": "删除 /tmp/all", "approved": False, "feedback": "", "result": ""}, config=cfg)
// 人工审查后修改 State
// app.update_state(cfg, values={"action": "列出 /tmp/all 文件而非删除"})
// 继续执行
// result = app.invoke(None, config=cfg)
return result;
// ── 模式3：对话模式 — Agent 主动向人提问 ──
// def ask_human(state: HitlState) -> dict:
/** docstring */
// question = "我不确定你想删除哪个目录，请指定："
// 实际项目中通过 UI 或 chat 将问题发给用户
return {"result": question, "feedback": "等待用户回答"};
function execute_dialog_mode() {
/** docstring */
// wg = StateGraph(HitlState)
// wg.add_node("risky", risky_action)
// wg.add_node("ask", ask_human)
// wg.add_edge(START, "risky")
// 条件边：不确定时跳到 ask 节点
// wg.add_conditional_edges("risky", lambda s: "ask" if not s.get("approved") else END,
// {"ask": "ask", END: END})
// wg.add_edge("ask", END)
return wg.compile(checkpointer=MemorySaver());
}
package main
import (
"fmt"
"os"
"os/exec"
"strings"
)
// from langgraph.graph import StateGraph, END, START
// from langgraph.checkpoint.memory import MemorySaver
// from typing import TypedDict
// HitlState - CLI Agent class
type HitlState struct {
// Python: action: str
// Python: approved: bool
// Python: feedback: str
// Python: result: str
// ── 模式1：审批模式 — 执行前等待人工批准 ──
// Python: def risky_action(state: HitlState) -> dict:
return {"result": f"已执行操作: {state['action']}"}
func execute_approval_mode() {
// Python: wg = StateGraph(HitlState)
// Python: wg.add_node("risky", risky_action)
// Python: wg.add_edge(START, "risky")
// Python: wg.add_edge("risky", END)
// Python: app = wg.compile(
// Python: checkpointer=MemorySaver(),
// Python: interrupt_before=["risky"] # 关键：执行前暂停
// Python: )
return app
// ── 模式2：纠正模式 — 人工修改 State 后继续 ──
func execute_correction_mode() {
// Python: cfg = {"configurable": {"thread_id": thread_id}}
// 第一次调用：执行到暂停点
// Python: state = app.invoke({"action": "删除 /tmp/all", "approved": False, "feedback": "", "result": ""}, config=cfg)
// 人工审查后修改 State
// Python: app.update_state(cfg, values={"action": "列出 /tmp/all 文件而非删除"})
// 继续执行
// Python: result = app.invoke(None, config=cfg)
return result
// ── 模式3：对话模式 — Agent 主动向人提问 ──
// Python: def ask_human(state: HitlState) -> dict:
// Python: question = "我不确定你想删除哪个目录，请指定："
// 实际项目中通过 UI 或 chat 将问题发给用户
return {"result": question, "feedback": "等待用户回答"}
func execute_dialog_mode() {
// Python: wg = StateGraph(HitlState)
// Python: wg.add_node("risky", risky_action)
// Python: wg.add_node("ask", ask_human)
// Python: wg.add_edge(START, "risky")
// 条件边：不确定时跳到 ask 节点
// Python: wg.add_conditional_edges("risky", lambda s: "ask" if not s.get("approved") else END,
// Python: {"ask": "ask", END: END})
// Python: wg.add_edge("ask", END)
return wg.compile(checkpointer=MemorySaver())
}
import java.util.*;
import java.util.concurrent.*;
import java.util.regex.*;
import java.io.*;
// from langgraph.graph import StateGraph, END, START
// from langgraph.checkpoint.memory import MemorySaver
// from typing import TypedDict
public class HitlState {
// Python: action: str
// Python: approved: bool
// Python: feedback: str
// Python: result: str
// ── 模式1：审批模式 — 执行前等待人工批准 ──
// Python: def risky_action(state: HitlState) -> dict:
return {"result": f"已执行操作: {state['action']}"};
public static void execute_approval_mode() {
// Python: wg = StateGraph(HitlState)
// Python: wg.add_node("risky", risky_action)
// Python: wg.add_edge(START, "risky")
// Python: wg.add_edge("risky", END)
// Python: app = wg.compile(
// Python: checkpointer=MemorySaver(),
// Python: interrupt_before=["risky"] # 关键：执行前暂停
// Python: )
return app;
// ── 模式2：纠正模式 — 人工修改 State 后继续 ──
public static void execute_correction_mode() {
// Python: cfg = {"configurable": {"thread_id": thread_id}}
// 第一次调用：执行到暂停点
// Python: state = app.invoke({"action": "删除 /tmp/all", "approved": False, "feedback": "", "result": ""}, config=cfg)
// 人工审查后修改 State
// Python: app.update_state(cfg, values={"action": "列出 /tmp/all 文件而非删除"})
// 继续执行
// Python: result = app.invoke(None, config=cfg)
return result;
// ── 模式3：对话模式 — Agent 主动向人提问 ──
// Python: def ask_human(state: HitlState) -> dict:
// Python: question = "我不确定你想删除哪个目录，请指定："
// 实际项目中通过 UI 或 chat 将问题发给用户
return {"result": question, "feedback": "等待用户回答"};
public static void execute_dialog_mode() {
// Python: wg = StateGraph(HitlState)
// Python: wg.add_node("risky", risky_action)
// Python: wg.add_node("ask", ask_human)
// Python: wg.add_edge(START, "risky")
// 条件边：不确定时跳到 ask 节点
// Python: wg.add_conditional_edges("risky", lambda s: "ask" if not s.get("approved") else END,
// Python: {"ask": "ask", END: END})
// Python: wg.add_edge("ask", END)
return wg.compile(checkpointer=MemorySaver());
}
}
三种模式各有适用场景：审批模式适合高风险操作（如删除数据、发送邮件），纠正模式适合 Agent 理解有偏差时人工修正，对话模式适合 Agent 信息不足需要向人补充询问。生产项目中可以组合使用，比如先对话模式收集信息，再审批模式确认执行。
Streaming：流式输出实时反馈
用户体验角度，长时间运行的 Agent 必须提供实时反馈，否则用户会以为程序卡住了。LangGraph 原生支持 Streaming，可以流式输出每个节点的结果、State 变化以及 LLM 的 token 级流式输出。
from langgraph.graph import StateGraph, END, START
from langgraph.checkpoint.memory import MemorySaver
from typing import TypedDict, Annotated
import operator
class StreamState(TypedDict):
messages: Annotated[list, operator.add]
current_step: str
def thinking_node(state: StreamState) -> dict:
"""模拟 LLM 思考过程"""
return {"messages": [{"role": "ai", "content": "正在分析问题..."}], "current_step": "thinking"}
def searching_node(state: StreamState) -> dict:
"""模拟搜索过程"""
return {"messages": [{"role": "tool", "content": "找到 3 条相关结果"}], "current_step": "searching"}
def responding_node(state: StreamState) -> dict:
"""生成最终回复"""
return {"messages": [{"role": "ai", "content": "基于搜索结果，答案是..."}], "current_step": "responding"}
# 构建图
wg = StateGraph(StreamState)
wg.add_node("think", thinking_node)
wg.add_node("search", searching_node)
wg.add_node("respond", responding_node)
wg.add_edge(START, "think")
wg.add_edge("think", "search")
wg.add_edge("search", "respond")
wg.add_edge("respond", END)
app = wg.compile(checkpointer=MemorySaver())
# ── 方式1：stream 模式 — 逐节点输出 ──
print("=== stream 模式 ===")
for event in app.stream(
{"messages": [{"role": "user", "content": "什么是 LangGraph？"}], "current_step": ""},
config={"configurable": {"thread_id": "stream-001"}}
):
node_name = list(event.keys())[0]
print(f"[{node_name}] → {event[node_name]}")
# ── 方式2：stream_values 模式 — 每步输出完整 State ──
print("\n=== stream_values 模式 ===")
for state in app.stream_values(
{"messages": [{"role": "user", "content": "什么是 LangGraph？"}], "current_step": ""},
config={"configurable": {"thread_id": "stream-002"}}
):
print(f"current_step={state.get('current_step')}, messages_count={len(state.get('messages', []))}")
# ── 方式3：stream_events 模式 — 细粒度事件流（含 LLM token） ──
print("\n=== stream_events 模式 ===")
async for event in app.astream_events(
{"messages": [{"role": "user", "content": "什么是 LangGraph？"}], "current_step": ""},
config={"configurable": {"thread_id": "stream-003"}},
version="v2"
):
if event["event"] == "on_chain_start":
print(f"节点开始: {event['name']}")
elif event["event"] == "on_chain_end":
print(f"节点完成: {event['name']}, 输出: {event.get('data', {}).get('output', '')[:80]}")
import {StateGraph, END, START} from 'langgraph.graph';
import {MemorySaver} from 'langgraph.checkpoint.memory';
// TypeScript has built-in types, no import needed for TypedDict, Annotated
import * as operator from 'operator';
class StreamState {
// messages: Annotated[list, operator.add]
// current_step: str
// def thinking_node(state: StreamState) -> dict:
/** docstring */
return {"messages": [{"role": "ai", "content": "正在分析问题..."}], "current_step": "thinking"};
// def searching_node(state: StreamState) -> dict:
/** docstring */
return {"messages": [{"role": "tool", "content": "找到 3 条相关结果"}], "current_step": "searching"};
// def responding_node(state: StreamState) -> dict:
/** docstring */
return {"messages": [{"role": "ai", "content": "基于搜索结果，答案是..."}], "current_step": "responding"};
// 构建图
// wg = StateGraph(StreamState)
// wg.add_node("think", thinking_node)
// wg.add_node("search", searching_node)
// wg.add_node("respond", responding_node)
// wg.add_edge(START, "think")
// wg.add_edge("think", "search")
// wg.add_edge("search", "respond")
// wg.add_edge("respond", END)
// app = wg.compile(checkpointer=MemorySaver())
// ── 方式1：stream 模式 — 逐节点输出 ──
console.log("=== stream 模式 ===");
// for event in app.stream(
// {"messages": [{"role": "user", "content": "什么是 LangGraph？"}], "current_step": ""},
// config = {"configurable": {"thread_id": "stream-001"}}
// ):
// node_name = list(event.keys())[0]
console.log(`[${$1}] → {event[node_name]}`);
// ── 方式2：stream_values 模式 — 每步输出完整 State ──
console.log("\n=== stream_values 模式 ===");
// for state in app.stream_values(
// {"messages": [{"role": "user", "content": "什么是 LangGraph？"}], "current_step": ""},
// config = {"configurable": {"thread_id": "stream-002"}}
// ):
console.log(`current_step={state.get('current_step')}, messages_count={len(state.get('messages', []))}`);
// ── 方式3：stream_events 模式 — 细粒度事件流（含 LLM token） ──
console.log("\n=== stream_events 模式 ===");
// async for event in app.astream_events(
// {"messages": [{"role": "user", "content": "什么是 LangGraph？"}], "current_step": ""},
// config = {"configurable": {"thread_id": "stream-003"}},
// version = "v2"
// ):
if (event["event"] == "on_chain_start") {
console.log(`节点开始: {event['name']}`);
} else if (event["event"] == "on_chain_end") {
console.log(`节点完成: {event['name']}, 输出: {event.get('data', {}).get('output', '')[:80]}`);
}
package main
import (
"fmt"
"os"
"os/exec"
"strings"
)
// from langgraph.graph import StateGraph, END, START
// from langgraph.checkpoint.memory import MemorySaver
// from typing import TypedDict, Annotated
// import operator
// StreamState - CLI Agent class
type StreamState struct {
// Python: messages: Annotated[list, operator.add]
// Python: current_step: str
// Python: def thinking_node(state: StreamState) -> dict:
return {"messages": [{"role": "ai", "content": "正在分析问题..."}], "current_step": "thinking"}
// Python: def searching_node(state: StreamState) -> dict:
return {"messages": [{"role": "tool", "content": "找到 3 条相关结果"}], "current_step": "searching"}
// Python: def responding_node(state: StreamState) -> dict:
return {"messages": [{"role": "ai", "content": "基于搜索结果，答案是..."}], "current_step": "responding"}
// 构建图
// Python: wg = StateGraph(StreamState)
// Python: wg.add_node("think", thinking_node)
// Python: wg.add_node("search", searching_node)
// Python: wg.add_node("respond", responding_node)
// Python: wg.add_edge(START, "think")
// Python: wg.add_edge("think", "search")
// Python: wg.add_edge("search", "respond")
// Python: wg.add_edge("respond", END)
// Python: app = wg.compile(checkpointer=MemorySaver())
// ── 方式1：stream 模式 — 逐节点输出 ──
fmt.Println("=== stream 模式 ===")
// Python: for event in app.stream(
// Python: {"messages": [{"role": "user", "content": "什么是 LangGraph？"}], "current_step": ""},
// Python: config={"configurable": {"thread_id": "stream-001"}}
// Python: ):
// Python: node_name = list(event.keys())[0]
fmt.Println(f"[{node_name}] → {event[node_name]}")
// ── 方式2：stream_values 模式 — 每步输出完整 State ──
fmt.Println("\n=== stream_values 模式 ===")
// Python: for state in app.stream_values(
// Python: {"messages": [{"role": "user", "content": "什么是 LangGraph？"}], "current_step": ""},
// Python: config={"configurable": {"thread_id": "stream-002"}}
// Python: ):
fmt.Println(f"current_step={state.get('current_step')}, messages_count={len(state.get('messages', []))}")
// ── 方式3：stream_events 模式 — 细粒度事件流（含 LLM token） ──
fmt.Println("\n=== stream_events 模式 ===")
// Python: async for event in app.astream_events(
// Python: {"messages": [{"role": "user", "content": "什么是 LangGraph？"}], "current_step": ""},
// Python: config={"configurable": {"thread_id": "stream-003"}},
// Python: version="v2"
// Python: ):
if event["event"] == "on_chain_start" {
fmt.Println(f"节点开始: {event['name']}")
} else if event["event"] == "on_chain_end" {
fmt.Println(f"节点完成: {event['name']}, 输出: {event.get('data', {}).get('output', '')[:80]}")
}
import java.util.*;
import java.util.concurrent.*;
import java.util.regex.*;
import java.io.*;
// from langgraph.graph import StateGraph, END, START
// from langgraph.checkpoint.memory import MemorySaver
// from typing import TypedDict, Annotated
// import operator
public class StreamState {
// Python: messages: Annotated[list, operator.add]
// Python: current_step: str
// Python: def thinking_node(state: StreamState) -> dict:
return {"messages": [{"role": "ai", "content": "正在分析问题..."}], "current_step": "thinking"};
// Python: def searching_node(state: StreamState) -> dict:
return {"messages": [{"role": "tool", "content": "找到 3 条相关结果"}], "current_step": "searching"};
// Python: def responding_node(state: StreamState) -> dict:
return {"messages": [{"role": "ai", "content": "基于搜索结果，答案是..."}], "current_step": "responding"};
// 构建图
// Python: wg = StateGraph(StreamState)
// Python: wg.add_node("think", thinking_node)
// Python: wg.add_node("search", searching_node)
// Python: wg.add_node("respond", responding_node)
// Python: wg.add_edge(START, "think")
// Python: wg.add_edge("think", "search")
// Python: wg.add_edge("search", "respond")
// Python: wg.add_edge("respond", END)
// Python: app = wg.compile(checkpointer=MemorySaver())
// ── 方式1：stream 模式 — 逐节点输出 ──
System.out.println("=== stream 模式 ===");
// Python: for event in app.stream(
// Python: {"messages": [{"role": "user", "content": "什么是 LangGraph？"}], "current_step": ""},
// Python: config={"configurable": {"thread_id": "stream-001"}}
// Python: ):
// Python: node_name = list(event.keys())[0]
System.out.println(String.format("$1"));
// ── 方式2：stream_values 模式 — 每步输出完整 State ──
System.out.println("\n=== stream_values 模式 ===");
// Python: for state in app.stream_values(
// Python: {"messages": [{"role": "user", "content": "什么是 LangGraph？"}], "current_step": ""},
// Python: config={"configurable": {"thread_id": "stream-002"}}
// Python: ):
System.out.println(String.format("$1"));
// ── 方式3：stream_events 模式 — 细粒度事件流（含 LLM token） ──
System.out.println("\n=== stream_events 模式 ===");
// Python: async for event in app.astream_events(
// Python: {"messages": [{"role": "user", "content": "什么是 LangGraph？"}], "current_step": ""},
// Python: config={"configurable": {"thread_id": "stream-003"}},
// Python: version="v2"
// Python: ):
if (event["event"] == "on_chain_start") {
System.out.println(String.format("$1"));
} else if (event["event"] == "on_chain_end") {
System.out.println(String.format("$1"));
}
}
三种 Streaming 方式从粗到细各有用途：stream() 以节点为单位输出，适合展示「正在执行哪一步」的进度条；stream_values() 每步输出完整 State，适合调试和监控状态变化；astream_events() 是最细粒度的事件流，能捕获 LLM 的 token 级输出，适合实现打字机效果的实时回复。在生产项目中，推荐前端用 astream_events 做 token 级展示，后端用 stream 做节点级日志记录。
📋 八股总结 — 面试高频考点
Q1: LangGraph 的核心思想是什么？与 LangChain Agent 有什么区别？
核心思想：把 Agent 工作流建模为有向图。节点 = 执行步骤，边 = 跳转逻辑。
区别：LangChain Agent 是黑盒（LLM 自主决定流程），难调试、难干预。LangGraph 是白盒（图结构显式定义），每步可检查 State、支持人工干预、支持并行。
简单场景用 LangChain Agent，复杂生产场景用 LangGraph。
Q2: LangGraph 的五个核心概念是什么？
① State（状态）：全局共享状态，所有节点可读写。用 TypedDict 定义。
② Node（节点）：执行单元，一个函数，接收 State 返回更新后的 State。
③ Edge（边）：节点间连接，普通边=固定跳转。
④ Conditional Edge（条件边）：根据 State 动态路由，类似 if-else。
⑤ Checkpoint（检查点）：每步自动保存 State 快照，支持时间旅行、人工干预、错误恢复。
Q3: LangGraph 如何实现 Human-in-the-Loop？
通过 Checkpoint + interrupt_before/interrupt_after 实现：
① 编译图时指定 interrupt_before=["tools"]，表示在 tools 节点前暂停
② Agent 执行到暂停点时停止，等待人工操作
③ 人工可以：直接继续（批准）、修改 State 后继续（纠正）、终止执行（拒绝）
④ 通过 thread_id 关联同一次执行的状态快照，支持从任意点恢复
Q4: LangGraph 中 State 的 Annotated[list, operator.add] 是什么意思？
Annotated[list, operator.add] 定义了 State 字段的合并策略。
operator.add 表示"追加"语义：当节点返回新的 messages 时，新消息追加到已有列表，而不是覆盖。
这对消息历史很重要——每个节点都应该看到完整的对话历史，而不是只看到上一步的结果。
Q5: LangGraph 如何实现 ReAct 循环？
通过图结构：
① agent 节点（调用 LLM 推理）
② tools 节点（执行工具）
③ 条件边 should_continue：检查最后一条消息有无 tool_calls
④ agent → (条件) → tools → agent → ... 形成循环
⑤ 当 LLM 不再要求调用工具时，条件边路由到 END，循环结束
ReAct 的 Thought-Action-Observation 循环在 LangGraph 中体现为图中的环（cycle）。
Q6: LangGraph 中 StateGraph 的核心数据结构是什么？为什么用 TypedDict？
核心数据结构是 StateGraph，它以 TypedDict 作为 State 的类型定义。
使用 TypedDict 的原因有三点：① 类型安全，IDE 可以自动补全字段名和类型，减少拼写错误；② 合并策略声明，通过 Annotated[list, operator.add] 可以指定字段的合并方式（追加 vs 覆盖）；③ 可序列化，TypedDict 本质是 dict，可以直接序列化为 JSON 保存到 Checkpoint 中。
StateGraph 在编译时会根据 TypedDict 的字段定义自动处理节点返回值与现有 State 的合并逻辑，无需手动编写 merge 函数。
Q7: ConditionEdge 和普通 Edge 的区别是什么？什么场景下用 ConditionEdge？
普通 Edge：固定跳转，A 节点执行完必定跳到 B 节点。语法：graph.add_edge("A", "B")。
Conditional Edge：动态路由，根据当前 State 的内容决定下一个节点。语法：graph.add_conditional_edges("A", router_fn, {"path1": "node1", "path2": "node2"})。
使用场景：任何需要根据运行时状态做分支的地方都应该用 ConditionEdge。典型场景包括：ReAct 循环中判断是否继续调用工具、多工具选择（根据 LLM 输出决定调用哪个工具）、错误处理（根据错误类型决定重试还是终止）、质量审查（根据评分决定通过还是重试）。
Q8: Checkpoint 在 LangGraph 中的作用是什么？MemorySaver 和其他 Checkpointer 有什么区别？
Checkpoint 作用：在每个节点执行后自动保存 State 快照，实现三大功能——时间旅行（回到任意历史状态）、人工干预（暂停在指定节点等待人工操作后继续）、错误恢复（从失败点重试而非从头开始）。
MemorySaver：内存级 Checkpointer，数据存在进程内存中，重启后丢失。适合开发调试和单机部署。
SqliteSaver / PostgresSaver：持久化到数据库，重启后数据仍在。适合生产环境，支持多实例共享状态。
选型建议：开发用 MemorySaver，生产用 PostgresSaver。切换只需改一行 compile(checkpointer=xxx)，图定义完全不变。
Q9: LangGraph 中 Human-in-the-Loop 的三种模式分别是什么？如何实现？
模式1：审批模式。编译时设置 interrupt_before=["node_name"]，在指定节点前自动暂停，等待人工批准后继续。适合高风险操作。
模式2：纠正模式。暂停后通过 app.update_state(config, values={...}) 修改 State 内容，再调用 app.invoke(None, config) 继续。适合 Agent 理解偏差时人工修正。
模式3：对话模式。在条件边中判断是否需要人工输入，跳转到专门的人机交互节点等待用户回答。适合 Agent 信息不足需要补充询问的场景。
三种模式可以组合使用，通过 thread_id 关联同一次执行的状态快照。
Q10: 项目选型时，LangGraph 和 LangChain Agent 怎么选？给出决策标准。
选 LangChain Agent 的场景：① 快速原型验证，不在乎流程控制；② 简单的单步或少步 Agent，不需要循环和并行；③ 不需要人工干预和状态持久化；④ 团队已经在使用 LangChain 生态，不想引入新依赖。
选 LangGraph 的场景：① 需要精细控制 Agent 每一步的流程；② 需要人工干预（审批、纠正、对话）；③ 需要并行执行多个节点；④ 需要状态持久化和错误恢复；⑤ 生产环境部署，需要可调试性和可观测性；⑥ 复杂工作流，涉及子图、多 Agent 协作。
一句话标准：能用 LangChain Agent 在 10 分钟跑通的简单场景就用它，但凡需要「控制力」就上 LangGraph。
Q11: LangGraph 的节点和边，和传统工作流有什么区别？
核心区别 = 灵活性 + 循环能力：
传统工作流：节点和跳转边都是固定写死的，一步错就全盘卡壳，也不支持循环执行。适用于流程确定、场景固定的简单业务。
LangGraph：
① 支持条件边——下一个执行节点不是固定的，由大模型的推理结果动态决定。类似 if-else 但决策者是 LLM 而非硬编码逻辑。
② 支持循环执行——Agent 可以自主反复试错、迭代优化，直到完成任务。这是 Agent 能自主完成复杂任务的核心原因。
一句话话术：传统工作流是确定性流水线，LangGraph是不确定性自主推理链。条件边让 Agent 能"自主决策"下一步，循环执行让 Agent 能"试错优化"直到完成。
第15章 LangGraph与状态机
http://www.clxhxhhr.top/posts/718/ 评论
0 条
还没有评论，先写一条吧。