LangGraph 是 LangChain 生态中用于构建状态机驱动的多智能体工作流的库，核心优势是支持循环、分支、条件判断和状态管理，特别适合复杂的多步骤任务（如数据分析、报告生成、多智能体协作等）。

以下是 LangGraph 的核心概念、基本使用步骤和实战示例，帮助你快速上手：

一、核心概念

在使用 LangGraph 前，先理解 3 个关键组件：

1. 状态（State）：工作流的“数据容器”，存储所有中间结果（如用户输入、工具输出、智能体回复等），贯穿整个流程。
2. 节点（Node）：工作流的“执行单元”，对应一个具体操作（如调用 LLM、执行工具、数据处理等），输入是状态，输出会更新状态。
3. 边（Edge）：节点之间的“连接规则”，定义流程走向（如固定下一步、条件分支、循环）。
4. 图（Graph）：由节点和边组成的完整工作流，通过 Graph 类定义并运行。

二、环境准备

首先安装依赖（需 Python 3.8+）：

pip install langgraph langchain-openai  # 核心依赖（LangGraph + OpenAI LLM）
# 如需其他 LLM（如 Anthropic、本地化模型），安装对应适配器：
# pip install langchain-anthropic langchain-ollama

三、基本使用步骤（四步走）

步骤 1：定义状态（State）

状态是 LangGraph 的核心，需明确存储哪些数据。推荐使用 TypedDict 定义结构化状态（类型提示更清晰）。

示例：定义一个“问答+工具调用”的状态，包含用户问题、LLM 思考过程、工具结果、最终答案：

from typing import TypedDict, Optional

class QAState(TypedDict):
    user_query: str  # 用户输入的问题（必填）
    llm_thought: Optional[str]  # LLM 的思考过程（可选，初始为 None）
    tool_result: Optional[str]  # 工具调用结果（可选）
    final_answer: Optional[str]  # 最终回复（可选）

步骤 2：定义节点（Node）

节点是具体的执行函数，输入是状态（State），输出是更新后的状态片段（会合并到全局状态）。

常见节点类型：

• LLM 思考节点：调用大模型分析问题、决定是否需要工具。
• 工具执行节点：调用外部工具（如计算器、搜索引擎）获取结果。
• 结果整理节点：将工具结果或 LLM 回复整理为最终答案。

示例：定义 3 个节点：

from langchain_openai import ChatOpenAI

# 初始化 LLM（需设置 OPENAI_API_KEY 环境变量）
llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0)

# 节点 1：LLM 思考（判断是否需要调用工具）
def llm_thought_node(state: QAState) -> dict:
    user_query = state["user_query"]
    # 让 LLM 分析问题，判断是否需要工具（如数学计算、实时数据）
    prompt = f"""
    你需要分析用户问题：{user_query}
    1. 若问题需要数学计算、实时数据或外部信息，回复“需要工具”；
    2. 若问题可直接回答，回复“无需工具”。
    只返回结果，不要额外内容。
    """
    thought = llm.invoke(prompt).content
    return {"llm_thought": thought}  # 仅返回需要更新的状态字段

# 节点 2：工具执行（示例：简单的数学计算工具）
def tool_execution_node(state: QAState) -> dict:
    user_query = state["user_query"]
    # 模拟工具调用：这里用 eval 简单计算（实际场景需用安全的工具库）
    try:
        # 提取数学表达式（简化处理，仅支持简单计算）
        expr = user_query.replace("=", "").strip()
        result = eval(expr)  # 注意：生产环境禁用 eval，用 sympy 等安全库
        return {"tool_result": f"计算结果：{expr} = {result}"}
    except Exception as e:
        return {"tool_result": f"工具调用失败：{str(e)}"}

# 节点 3：生成最终答案
def final_answer_node(state: QAState) -> dict:
    user_query = state["user_query"]
    llm_thought = state["llm_thought"]
    tool_result = state["tool_result"]
    
    # 整合信息生成最终回复
    prompt = f"""
    用户问题：{user_query}
    思考过程：{llm_thought}
    工具结果（若有）：{tool_result or '无'}
    请基于以上信息，生成自然、准确的最终回答。
    """
    final_answer = llm.invoke(prompt).content
    return {"final_answer": final_answer}

步骤 3：定义图（Graph）与边（Edge）

通过 langgraph.graph.Graph 类组装节点和边，核心是定义流程走向规则。

3.1 基础流程（无分支）

如果流程是固定顺序（如：思考 → 工具 → 最终答案），直接用 add_edge 串联：

from langgraph.graph import Graph

# 1. 初始化图（指定状态类型）
graph = Graph(QAState)

# 2. 添加节点（节点名称可自定义，与函数对应）
graph.add_node("llm_thought", llm_thought_node)
graph.add_node("tool_execution", tool_execution_node)
graph.add_node("final_answer", final_answer_node)

# 3. 定义边（固定顺序：思考 → 工具 → 最终答案）
graph.add_edge("llm_thought", "tool_execution")
graph.add_edge("tool_execution", "final_answer")

# 4. 设置入口节点（流程开始的第一个节点）
graph.set_entry_point("llm_thought")

# 5. 设置结束节点（流程终止的节点）
graph.set_finish_point("final_answer")

# 6. 编译图（生成可运行的工作流）
app = graph.compile()

3.2 条件分支（核心功能）

实际场景中常需要根据节点输出判断下一步（如：LLM 认为“无需工具”则跳过工具节点）。

通过 add_conditional_edges 定义分支规则，规则函数需返回下一个节点的名称：

# 重新定义图，支持条件分支
graph = Graph(QAState)

# 添加节点（同上）
graph.add_node("llm_thought", llm_thought_node)
graph.add_node("tool_execution", tool_execution_node)
graph.add_node("final_answer", final_answer_node)

# 定义条件分支规则：根据 llm_thought 的结果决定下一步
def decide_next_node(state: QAState) -> str:
    thought = state["llm_thought"]
    if "需要工具" in thought:
        return "tool_execution"  # 需工具 → 执行工具
    else:
        return "final_answer"  # 无需工具 → 直接生成最终答案

# 添加条件边（从 llm_thought 节点出发，根据规则选择下一步）
graph.add_conditional_edges(
    source="llm_thought",  # 源节点
    condition=decide_next_node,  # 分支规则函数
    # 可选：显式指定可能的目标节点（增强类型检查）
    dests=["tool_execution", "final_answer"]
)

# 工具执行后，固定走向最终答案
graph.add_edge("tool_execution", "final_answer")

# 设置入口和结束节点
graph.set_entry_point("llm_thought")
graph.set_finish_point("final_answer")

# 编译图
app = graph.compile()

步骤 4：运行工作流

通过 app.invoke() 传入初始状态（仅需包含必填字段），执行工作流并获取结果：

# 示例 1：需要工具的问题（数学计算）
input_state1 = {"user_query": "100 + 200 * 3 = ?"}
result1 = app.invoke(input_state1)
print("最终答案 1：", result1["final_answer"])

# 示例 2：无需工具的问题（常识问答）
input_state2 = {"user_query": "地球的半径大约是多少公里？"}
result2 = app.invoke(input_state2)
print("最终答案 2：", result2["final_answer"])

输出示例：

最终答案 1： 根据计算，100 + 200 * 3 的结果为 700（先计算乘法 200×3=600，再计算加法 100+600=700）。
最终答案 2： 地球的平均半径约为 6371 公里（赤道半径约 6378 公里，极半径约 6357 公里）。

四、进阶功能（可选）

1. 可视化工作流

LangGraph 支持用 Mermaid 语法可视化图结构，需安装 pygraphviz 或 matplotlib：

pip install pygraphviz  # 或 pip install matplotlib

可视化代码：

# 生成 Mermaid 图表（可复制到 https://mermaid-js.github.io/mermaid-live-editor/ 查看）
print(app.get_graph().draw_mermaid())

# 直接保存为图片（需 pygraphviz）
app.get_graph().draw("workflow.png", format="png")

2. 循环逻辑（如多轮工具调用）

如果需要反复调用工具（如 LLM 需多次补充信息），可让节点返回自身作为下一步：

# 示例：工具调用失败时，重新执行工具节点（循环）
def tool_execution_node_with_retry(state: QAState) -> dict:
    # （同上的工具执行逻辑）
    ...

# 分支规则：工具调用失败则重试，成功则进入最终答案
def decide_after_tool(state: QAState) -> str:
    tool_result = state["tool_result"]
    if "失败" in tool_result:
        return "tool_execution"  # 循环重试
    else:
        return "final_answer"

# 添加循环边
graph.add_conditional_edges("tool_execution", decide_after_tool, ["tool_execution", "final_answer"])

3. 多智能体协作

LangGraph 可实现多智能体分工（如“分析师”→“审核师”→“输出员”），每个智能体作为一个节点：

# 示例：多智能体节点
def analyst_node(state: QAState) -> dict:
    # 分析师：分析问题并生成数据需求
    ...

def reviewer_node(state: QAState) -> dict:
    # 审核师：验证分析师的结论是否准确
    ...

# 组装图：analyst → reviewer → final_answer
graph.add_node("analyst", analyst_node)
graph.add_node("reviewer", reviewer_node)
graph.add_edge("analyst", "reviewer")
graph.add_edge("reviewer", "final_answer")

五、关键注意事项

1. 状态设计：状态需包含所有中间数据，避免节点间传递额外参数，推荐用 TypedDict 结构化。
2. 节点纯度：节点函数应仅依赖输入状态，避免全局变量（确保可复现）。
3. 错误处理：工具调用、LLM 响应可能失败，需在节点中添加异常捕获，或通过循环重试处理。
4. 性能优化：复杂图可通过 langgraph.checkpoint 启用状态持久化（支持中断后恢复）。
5. 安全问题：避免使用 eval 等危险函数，工具调用优先使用 LangChain 的 Tool 类（如 CalculatorTool）。

六、总结

LangGraph 的核心是“状态驱动的流程编排”，通过以下步骤即可构建复杂工作流：

1. 定义状态（存储数据）→ 2. 实现节点（执行逻辑）→ 3. 组装图（定义流程）→ 4. 运行工作流。

它比传统的线性工作流更灵活，支持分支、循环、多智能体协作，是处理复杂任务（如数据分析、报告生成、客服对话）的强大工具。

如需更复杂的场景（如实时数据流、多模态输入、分布式执行），可参考 LangGraph 官方文档。