📚 AI 工程与智能体教程站

第 4 章 · LangGraph 概述与 StateGraph 入门 ​

本章目标：

  • 理解 LangGraph 的定位：低层级编排运行时，区别于 LangChain agents
  • 掌握 StateGraph 最小示例：节点、边、状态、编译与执行
  • 理解 MessagesState 与自定义状态
  • 区分 LangGraph Runtime 与 Deep Agents Harness 的关系

4.1 为什么需要 LangGraph ​

LangChain agents 提供开箱即用的架构（如 ReAct、Plan-and-Execute），但缺乏对执行流程的精细控制。LangGraph 填补了这个空白：

LangGraph 是一个低层级编排框架和运行时，用于构建、管理和部署长期运行的状态化 Agent。它允许你将确定性逻辑与 LLM 驱动的决策混合在同一个图中。

核心差异：

特性LangChain AgentsLangGraph
抽象层级高级低层级
流程控制黑盒循环显式图结构
状态持久化有限内置 Checkpointer
Human-in-the-loop不支持原生支持
调试困难LangSmith 可视化

4.2 最小 Hello World ​

python
from langgraph.graph import StateGraph, MessagesState, START, END

# 定义节点：简单的 LLM 调用
def mock_llm(state: MessagesState) -> dict:
    return {
        "messages": [
            {"role": "assistant", "content": "Hello from LangGraph!"}
        ]
    }

# 构建图
graph = StateGraph(MessagesState)
graph.add_node("llm", mock_llm)
graph.add_edge(START, "llm")
graph.add_edge("llm", END)
compiled = graph.compile()

# 执行
result = compiled.invoke({
    "messages": [{"role": "user", "content": "hi!"}]
})
print(result["messages"][-1]["content"])

4.3 StateGraph 核心概念 ​

python
from langgraph.graph import StateGraph, MessagesState, START, END
from typing import TypedDict

# 自定义状态：在 MessagesState 基础上添加 metadata
class AgentState(MessagesState):
    context: list[str]  # 检索到的上下文
    step_count: int     # 已执行步骤数
    final_answer: str   # 最终答案

# 各节点函数
def search_node(state: AgentState) -> dict:
    """检索相关文档"""
    return {
        "context": ["文档A内容", "文档B内容"],
        "step_count": state.get("step_count", 0) + 1
    }

def llm_node(state: AgentState) -> dict:
    """调用 LLM 生成回答"""
    context = " ".join(state.get("context", []))
    return {
        "messages": [{"role": "assistant", "content": f"基于上下文: {context}"}],
        "step_count": state["step_count"] + 1
    }

# 构建图
workflow = StateGraph(AgentState)
workflow.add_node("search", search_node)
workflow.add_node("llm", llm_node)
workflow.add_edge(START, "search")
workflow.add_edge("search", "llm")
workflow.add_edge("llm", END)

graph = workflow.compile()

4.4 与 LangChain Agents 的关系 ​

python
# LangGraph 可以与 LangChain 组件结合使用
from langchain.chat_models import ChatOpenAI
from langchain.tools import tool

@tool
def get_weather(city: str) -> str:
    """获取城市天气"""
    return f"{city} 今天晴朗，25°C"

# 使用 LangChain 的 LLM + LangGraph 的图编排
llm = ChatOpenAI(model="gpt-4o-mini")
# ... 将 LLM 调用封装为图节点

💡 更详细的工具调用见 第 5 章。

4.5 编译与执行模式 ​

python
# 同步执行
result = graph.invoke({"messages": [...]})

# 流式执行（见第 8 章）
for event in graph.stream({"messages": [...]}):
    print(event)

# 异步执行
async_result = await graph.ainvoke({"messages": [...]})

本章小结 ​

  • LangGraph 是低层级编排运行时，提供精细的图结构控制
  • StateGraph + MessagesState 是最小起步单元
  • 自定义状态继承 MessagesState 并扩展字段
  • 与 LangChain 组件可混合使用

🛠️ 动手实践 ​

  1. 实现一个最小 LangGraph 应用：两个节点（ greeted → end），输出 "Hello LangGraph!"
  2. 扩展 StateGraph，添加自定义状态字段（如 conversation_id）
  3. 使用 graph.stream() 实现流式输出