LangGraph 从入门到实战教程：本地部署与 Agent 框架选型深度对比

在 AI Agent 爆发的当下，很多开发者在实际业务落地时都遇到了同一个痛点：传统的 Agent 框架（如早期的 LangChain Agent）在处理复杂、长链路任务时，往往像一个不可控的黑盒。一旦陷入死循环或发生工具调用错误，开发者极难介入调试。

为了解决这个问题，LangChain 官方推出了 LangGraph。它放弃了简单的线性执行，转而采用图结构（Graph）和状态机（State Machine）的理念，将 Agent 的执行过程彻底白盒化。

本文将深度拆解 LangGraph 的官方文档核心概念，提供一份可直接在本地环境（包括对接 Ollama 等本地模型）跑通的实战教程，并在最后从底层逻辑上将其与 OpenClaw、CrewAI 进行横向对比，为你的技术选型提供事实依据。

一、 为什么传统的 Agent 框架会失效？

在自动化运维等对准确性要求极高的 IT 场景中，业务流绝对不能是“掷骰子”。

传统 Agent 的运行逻辑是：给定一个 Prompt 和一堆 Tools，大模型自己决定下一步干什么。这种高度依赖大模型推理能力的模式（ReAct 范式），在处理简单任务时很惊艳，但面临多步条件分支时，极易发生“幻觉”或偏离主线。

LangGraph 的破局点在于：状态（State）与路由（Routing）的显性化。

它允许开发者明确规定：

1. Agent 在任意时刻共享什么样的数据结构（State）。
2. 节点（Node）执行完毕后，根据什么条件走到下一个节点（Conditional Edges）。
3. 如何利用检查点（Checkpoints）实现多轮对话的记忆持久化和人类介入（Human-in-the-loop）。

二、 LangGraph 核心 API 与官方概念对齐

在写代码之前，必须准确理解 LangGraph 官方文档中的几个核心类：

• StateGraph：整个工作流的载体。你需要定义一个 Python TypedDict 或 Pydantic 类作为状态的 Schema 传给它。
• Reducers：在定义 State 时，某些字段（如消息列表）需要追加而不是覆盖。LangGraph 官方推荐使用 typing.Annotated 结合 operator.add（或自带的 add_messages）来实现。
• Nodes：图中的节点，本质上是接收当前 State 并返回更新后 State 的纯 Python 函数。
• Edges & Conditional Edges：控制流。普通边定义了强顺序执行，条件边则是一个返回下一个节点名称的路由函数。
• Checkpointer：状态持久化机制。通过引入 MemorySaver 或 SqliteSaver，可以将图的运行状态在特定节点挂起，甚至随时回滚。

三、 LangGraph 深度实战：构建带记忆与工具调用的自省 Agent

接下来，我们将不使用简单的“打招呼”Demo，而是严格按照官方实践，构建一个包含工具调用（Tool Calling）、本地模型对接、状态追加和持久化记忆的 Agent。

1. 环境与依赖准备

建议在独立的虚拟环境中进行，确保版本兼容性。

pip install langgraph langchain-openai langchain-community
# 如果需要使用 SQLite 做持久化
pip install langgraph-checkpoint-sqlite

2. 本地化大模型接入配置

许多开发者更倾向于使用本地部署的模型来保护数据隐私并降低成本。这里演示如何通过 langchain-openai 兼容接入本地运行的 Ollama 或聚合 API 站点。

import os
from langchain_openai import ChatOpenAI

# 方案 A：对接本地 Ollama 服务 (需提前启动 ollama run qwen2)
llm = ChatOpenAI(
    model="qwen2", 
    base_url="http://localhost:11434/v1", 
    api_key="ollama_local",
    temperature=0
)

# 方案 B：对接自建的 API 代理/聚合站
# llm = ChatOpenAI(
#     model="gpt-4o", 
#     base_url="https://api.yourdomain.com/v1", 
#     api_key="sk-xxxx"
# )

3. 定义全局状态 (State) 与 Reducer

官方文档强烈建议使用 MessagesState，或者自定义带有 add_messages 的状态字典，以确保多轮对话历史不被截断。

from typing import TypedDict, Annotated, Sequence
from langchain_core.messages import BaseMessage
from langgraph.graph.message import add_messages

class AgentState(TypedDict):
    # Annotated 与 add_messages 确保每次返回消息时，是追加到列表末尾，而不是覆盖整个列表
    messages: Annotated[Sequence[BaseMessage], add_messages]
    # 你可以在这里追加自定义状态，例如任务重试次数
    error_count: int

4. 编写核心节点与工具

我们将定义一个可以查询本地时间的工具，并将其绑定到 LLM。

from langchain_core.tools import tool
from langgraph.prebuilt import ToolNode
from datetime import datetime

@tool
def get_current_time(location: str) -> str:
    """获取当前系统的准确时间。"""
    return f"当前 {location} 的时间是: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}"

tools = [get_current_time]
# 将工具绑定到 LLM 上，使其具备工具调用能力
llm_with_tools = llm.bind_tools(tools)

# 节点 1：LLM 推理节点
def agent_node(state: AgentState):
    messages = state["messages"]
    # LLM 根据当前对话历史进行推理
    response = llm_with_tools.invoke(messages)
    # 返回的内容会自动通过 add_messages 追加到状态流中
    return {"messages": [response]}

# 节点 2：工具执行节点 (LangGraph 提供了现成的 ToolNode 简化开发)
tool_node = ToolNode(tools)

5. 定义条件路由 (Conditional Edge)

这是最关键的一环，决定了 Agent 发现需要调用工具时，如何跳转。

from langgraph.graph import END

def should_continue(state: AgentState) -> str:
    """检查 LLM 的最新输出中是否包含工具调用请求"""
    messages = state["messages"]
    last_message = messages[-1]
    
    # 如果大模型决定调用工具，则路由到 tools 节点
    if last_message.tool_calls:
        return "tools"
    # 否则直接结束
    return END

6. 编排并编译工作流 (带持久化记忆)

from langgraph.graph import StateGraph, START
from langgraph.checkpoint.memory import MemorySaver

# 初始化带有持久化功能的内存检查点
memory = MemorySaver()

workflow = StateGraph(AgentState)

# 注册节点
workflow.add_node("agent", agent_node)
workflow.add_node("tools", tool_node)

# 定义边
workflow.add_edge(START, "agent")

# 条件路由：从 agent 节点出发，根据 should_continue 的结果决定走向
workflow.add_conditional_edges(
    "agent",
    should_continue,
    {
        "tools": "tools",
        END: END
    }
)

# 工具执行完毕后，强制回流到 agent 节点继续推理
workflow.add_edge("tools", "agent")

# 编译成可执行应用，并挂载记忆模块
app = workflow.compile(checkpointer=memory)

7. 运行与验证

加入 thread_id 可以让我们跟踪不同用户的独立会话（Session）。

from langchain_core.messages import HumanMessage

# 配置线程 ID，用于记忆隔离
config = {"configurable": {"thread_id": "user_session_001"}}

print(">>> 第一次对话：提出问题")
input_message = HumanMessage(content="北京现在的准确时间是几点？")
for event in app.stream({"messages": [input_message]}, config, stream_mode="values"):
    last_msg = event["messages"][-1]
    last_msg.pretty_print()

print("\n>>> 第二次对话：测试记忆能力")
input_message2 = HumanMessage(content="我刚才问了你哪个城市的时间？")
for event in app.stream({"messages": [input_message2]}, config, stream_mode="values"):
    last_msg = event["messages"][-1]
    last_msg.pretty_print()

在这套架构下，你随时可以通过 app.get_state(config) 提取当前运行到哪一步、变量内容是什么，真正实现了 AI 业务流的白盒可控。

四、 框架架构级横向对比：LangGraph vs OpenClaw vs CrewAI

市面上 Agent 框架繁多，这三款代表了目前三种截然不同的设计哲学。以下基于底层实现逻辑进行客观对比。

1. 架构逻辑差异

2. 深度剖析：场景决定选型

LangGraph：严肃商业系统的“控制阀”

LangGraph 本质上并不是一个“帮你写 Prompt 的工具”，它是一个并发编程和工作流引擎。对于有多年 IT 研发经验、特别是做过核心系统或复杂微服务编排的开发者来说，LangGraph 的设计非常贴合工程直觉。

• 优势：极强的鲁棒性。在复杂的长文本处理或 API 链式调用中，大模型极其容易出错。LangGraph 允许你写死重试逻辑、设定兜底节点（Fallback Nodes），保障系统像传统程序一样稳定运行。
• 适用场景：企业级自动化处理流、代码生成与自修复系统、RAG（检索增强生成）的多重质量校验环节。

OpenClaw：AI 技能落地的“挂载舱”

如果你的核心痛点是“我有一个本地跑着的模型，我想让它立刻具备联网、查数据库、跑代码的能力”，OpenClaw 提供的架构非常高效。它更像是一个连接大模型与外部世界的标准化接口平台。

• 优势：生态友好，部署快。对于喜欢折腾 Docker 容器化部署、本地挂载 Ollama 玩转各种模型的开发者，OpenClaw 在构建单体全能 Agent（或轻量级协作）时，能省去大量编写底层工具链的时间。
• 适用场景：个人效率助手开发、需要大量与本地环境/服务器环境交互的运维 Agent、API 中转调用验证。

CrewAI：内容生态的“脑暴团队”

CrewAI 屏蔽了底层的图流转机制，其核心代码往往只有几行：定义 Agent（给它一个人设），定义 Task，然后把它们放进一个 Crew 里开始运行。

• 优势：极度符合人类直觉。它让非硬核程序员也能轻松构建多智能体系统。
• 致命弱点：过度依赖底层模型（如 GPT-4）的推理能力。在处理需要严密数据校验或确定性极高的任务时，极易发生“角色间互相吹捧但偏离目标”、“任务循环未输出可用结果”的现象。
• 适用场景：自媒体文章批量生成、多角度竞品分析报告撰写、小说/故事设定大纲推演等容错率高且偏创意类的任务。

结语

不要盲目追求“框架越新越好”或“多智能体一定比单智能体强”。

如果你的目标是输出一篇具有 SEO 价值的爆款文章或头脑风暴，CrewAI 能快速带来惊喜；如果你在折腾本地算力和模型，希望快速赋予模型实战技能，OpenClaw 值得一试；但如果你要写一套稳定运行在服务器上、每天处理成千上万真实业务数据的 AI 中枢，请老老实实啃下 LangGraph 的状态机设计。