LangGraph 状态机 Agent

为什么需要 LangGraph

上一章的 ReAct Agent 有一个根本性问题：它是无状态的循环，无法处理：

• 需要暂停等待人工确认的步骤（Human-in-the-loop）
• 需要在错误后从特定节点重试，而非从头开始
• 并行执行多个互不依赖的子任务
• 多个 Agent 互相协作（多智能体架构）
• 跨会话持久化状态（用户关闭浏览器后再回来继续）

LangGraph 将 Agent 的执行过程建模为有向图（Directed Graph），每个节点是一个处理步骤，边定义了流转逻辑，State 在节点间流转并积累。

构建第一个 LangGraph Agent

from typing import TypedDict, Annotated
from langchain_core.messages import BaseMessage, HumanMessage
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, END
from langgraph.graph.message import add_messages  # 消息追加 reducer
from langgraph.prebuilt import ToolNode
from langchain_core.tools import tool

# 1. 定义 State 类型
class AgentState(TypedDict):
    # Annotated[..., add_messages] 表示每次更新追加消息，而非覆盖
    messages: Annotated[list[BaseMessage], add_messages]

# 2. 定义工具
@tool
def get_weather(city: str) -> str:
    """获取指定城市的当前天气信息"""
    # 模拟数据
    data = {"北京": "晴，25°C", "上海": "多云，28°C", "深圳": "小雨，30°C"}
    return data.get(city, f"没有 {city} 的天气数据")

tools = [get_weather]

# 3. 定义节点函数
llm = ChatOpenAI(model="gpt-4o").bind_tools(tools)

def call_model(state: AgentState) -> AgentState:
    """模型节点：调用 LLM，可能发起工具调用"""
    response = llm.invoke(state["messages"])
    return {"messages": [response]}  # add_messages 会追加

def should_continue(state: AgentState) -> str:
    """条件边：判断是否需要继续执行工具"""
    last_msg = state["messages"][-1]
    # 如果有工具调用请求，跳转到 tools 节点；否则结束
    if last_msg.tool_calls:
        return "tools"
    return END  # END 是 LangGraph 的特殊结束节点

# 4. 构建图
workflow = StateGraph(AgentState)

# 添加节点
workflow.add_node("agent", call_model)
workflow.add_node("tools", ToolNode(tools))  # ToolNode 自动执行工具调用

# 设置入口节点
workflow.set_entry_point("agent")

# 添加条件边（从 agent 节点出发，根据 should_continue 决定走向）
workflow.add_conditional_edges("agent", should_continue)

# 工具节点执行后，无条件回到 agent 节点
workflow.add_edge("tools", "agent")

# 5. 编译图（可选加 Checkpointer）
app = workflow.compile()

# 6. 运行
result = app.invoke({"messages": [HumanMessage(content="北京和上海今天的天气怎么样？")]})
print(result["messages"][-1].content)

Human-in-the-Loop：人工干预机制

from langgraph.checkpoint.memory import MemorySaver
from langgraph.types import interrupt

# 使用内存 Checkpointer（生产用 PostgresSaver 或 RedisSaver）
memory = MemorySaver()

# 编译时加入 Checkpointer
app = workflow.compile(checkpointer=memory)

# 在工具节点中加入人工确认步骤：
def confirm_and_execute(state: AgentState):
    """工具节点：执行前请求人工确认"""
    last_msg = state["messages"][-1]
    for tool_call in last_msg.tool_calls:
        # interrupt() 会暂停执行，等待人工输入
        approval = interrupt({
            "type": "confirm_tool_call",
            "tool": tool_call["name"],
            "args": tool_call["args"]
        })
        if not approval:
            return {"messages": [...拒绝消息...]}
    # 批准后执行工具
    return ToolNode(tools).invoke(state)