LangGraph 的**核心概念、基本使用步骤和实战示例**

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, Optionalclass 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).contentreturn {"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).contentreturn {"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 官方文档。